syz-cluster: better handle SeriesProcessor restarts

If the Loop() was restarted in between the moment we marked the session
as started in the DB and the moment we actually started the workflow,
there was no way back to the normal operation.

That was the reason of the sporadic TestProcessor failures we've seen in
the presubmit tests.

Handle this case in the code by just continuing the non-finished calls.

Closes #5776.

Author: a-nogikh

syz-cluster/controller/processor.go

@@ -154,11 +154,16 @@ func (sp *SeriesProcessor) handleSession(ctx context.Context, session *db.Sessio
 		switch status {
 		case workflow.StatusNotFound:
 			log.Printf("scheduling a workflow for %q", session.ID)
-			if err := sp.sessionRepo.Start(ctx, session.ID); err != nil {
+			err := sp.sessionRepo.Start(ctx, session.ID)
+			if err == db.ErrSessionAlreadyStarted {
+				// It may happen if the service was restarted right between the moment we updated the DB
+				// and actually started the workflow.
+				log.Printf("session %q was already marked as started, but there's no actual workflow", session.ID)
+			} else if err != nil {
 				app.Errorf("failed to mark session started: %v", err)
 				break
 			}
-			err := sp.workflows.Start(session.ID)
+			err = sp.workflows.Start(session.ID)
 			if err != nil {
 				app.Errorf("failed to start a workflow: %v", err)
 			}

syz-cluster/controller/processor_test.go

@@ -52,7 +52,9 @@ func TestProcessor(t *testing.T) {
 
 	awaitFinishedSessions(t, processor.seriesRepo, 2)
 
-	// Restart the loop.
+	// Emulate the service restart by aborting the loop.
+	// This may break the execution in arbitrary places, which actually resembles the environment in which the code
+	// will actually work. The bugs it triggers may be difficult to reproduce though.
 	cancel()
 	wg.Wait()