syz-cluster: clean up running steps of finished workflows

If the workflow step crashed or timed out, we used to have Running
status for such steps even though the session itself may be long
finished.

In order to prevent this inconsistency, on finishing each session go
through all remaining running steps and update their status to Error.

1 files changed, 12 insertions, 8 deletions

diff --git a/syz-cluster/pkg/db/session_test_repo.go b/syz-cluster/pkg/db/session_test_repo.go
index e5b923fd4..7043b8389 100644
--- a/syz-cluster/pkg/db/session_test_repo.go
+++ b/syz-cluster/pkg/db/session_test_repo.go
@@ -85,14 +85,7 @@ type FullSessionTest struct {
 }
 
 func (repo *SessionTestRepository) BySession(ctx context.Context, sessionID string) ([]*FullSessionTest, error) {
-	stmt := spanner.Statement{
-		SQL: "SELECT * FROM `SessionTests` WHERE `SessionID` = @session" +
-			" ORDER BY `UpdatedAt`",
-		Params: map[string]interface{}{"session": sessionID},
-	}
-	iter := repo.client.Single().Query(ctx, stmt)
-	defer iter.Stop()
-	list, err := readEntities[SessionTest](iter)
+	list, err := repo.BySessionRaw(ctx, sessionID)
 	if err != nil {
 		return nil, err
 	}
@@ -131,3 +124,14 @@ func (repo *SessionTestRepository) BySession(ctx context.Context, sessionID stri
 	}
 	return ret, nil
 }
+
+func (repo *SessionTestRepository) BySessionRaw(ctx context.Context, sessionID string) ([]*SessionTest, error) {
+	stmt := spanner.Statement{
+		SQL: "SELECT * FROM `SessionTests` WHERE `SessionID` = @session" +
+			" ORDER BY `UpdatedAt`",
+		Params: map[string]interface{}{"session": sessionID},
+	}
+	iter := repo.client.Single().Query(ctx, stmt)
+	defer iter.Stop()
+	return readEntities[SessionTest](iter)
+}