Fix unfinished task scheduling

cmd/media_repo/main.go

@@ -73,13 +73,6 @@ func main() {
 	logrus.Info("Starting up...")
 	runtime.RunStartupSequence()
 
-	logrus.Info("Checking background tasks...")
-	err = scanAndStartUnfinishedTasks()
-	if err != nil {
-		sentry.CaptureException(err)
-		logrus.Fatal(err)
-	}
-
 	logrus.Info("Starting recurring tasks...")
 	tasks.StartAll()
 

tasks/all.go

@@ -10,6 +10,8 @@ func StartAll() {
 	scheduleHourly(RecurringTaskPurgeRemoteMedia, task_runner.PurgeRemoteMedia)
 	scheduleHourly(RecurringTaskPurgeThumbnails, task_runner.PurgeThumbnails)
 	scheduleHourly(RecurringTaskPurgePreviews, task_runner.PurgePreviews)
+
+	scheduleUnfinished()
 }
 
 func StopAll() {

tasks/exec.go

@@ -56,6 +56,10 @@ func tryBeginTask(id int, recur bool) {
 }
 
 func beginTask(task *database.DbTask) {
+	if task.EndTs > 0 {
+		return // just skip it
+	}
+	// TODO: Worker group: https://github.com/turt2live/matrix-media-repo/issues/425
 	runnerCtx := rcontext.Initial().LogWithFields(logrus.Fields{"task_id": task.TaskId})
 	if task.Name == string(TaskDatastoreMigrate) {
 		go task_runner.DatastoreMigrate(runnerCtx, task)

tasks/schedule.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/getsentry/sentry-go"
 	"github.com/sirupsen/logrus"
 	"github.com/turt2live/matrix-media-repo/common/rcontext"
 	"github.com/turt2live/matrix-media-repo/database"
@@ -97,6 +98,23 @@ func stopRecurring() {
 	}
 }
 
+func scheduleUnfinished() {
+	if ids.GetMachineId() != 0 {
+		return // don't schedule here
+	}
+	ctx := rcontext.Initial().LogWithFields(logrus.Fields{"startup": true})
+	taskDb := database.GetInstance().Tasks.Prepare(ctx)
+	tasks, err := taskDb.GetAll(false)
+	if err != nil {
+		sentry.CaptureException(err)
+		ctx.Log.Fatal("Error getting unfinished tasks: ", err)
+		return
+	}
+	for _, task := range tasks {
+		beginTask(task)
+	}
+}
+
 func RunDatastoreMigration(ctx rcontext.RequestContext, sourceDsId string, targetDsId string, beforeTs int64) (*database.DbTask, error) {
 	return scheduleTask(ctx, TaskDatastoreMigrate, task_runner.DatastoreMigrateParams{
 		SourceDsId: sourceDsId,