up

2026-10-17 16:00:21 +07:00
parent ee0478ab16
commit 300ff484df
21 changed files with 1643 additions and 870 deletions
@@ -104,3 +104,27 @@ func (f exportFixture) start(t *testing.T, id string) *Attempt {
 	requireNoError(t, err)
 	return attempt
 }
+
+// releaseMigration exercises transactional schema changes without querying application data.
+func (f exportFixture) releaseMigration(t *testing.T, down bool) error {
+	t.Helper()
+	raw, err := os.ReadFile("../../../migrations/20261017120000_release_game_exports.sql")
+	requireNoError(t, err)
+	parts := strings.Split(string(raw), "-- +goose Down")
+	sql := parts[0]
+	if down {
+		sql = parts[1]
+	}
+	tx, err := f.pool.Begin(context.Background())
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(context.Background()) }()
+	if _, err := tx.Exec(
+		context.Background(),
+		sql,
+	); err != nil {
+		return err
+	}
+	return tx.Commit(context.Background())
+}
@@ -13,7 +13,7 @@ import (

 var (
 	ErrGameNotFound   = errors.New("игра не найдена")
-	ErrNotExportable  = errors.New("экспорт доступен только для игры в черновике без дат проведения и прогресса")
+	ErrExportChanged  = errors.New("экспорт игры изменился; обновите страницу и повторите действие")
 	ErrExportNotFound = errors.New("экспорт игры ещё не запущен")
 	ErrOwnershipLost  = errors.New("владение экспортом потеряно")
 )
@@ -45,79 +45,43 @@ func NewGameExportsRepo(pool *pgxpool.Pool) *GameExportsRepo {
 	}
 }

-// Start locks the game before inspecting progress, then saves all package data in one SQL snapshot.
+// Start serializes replacements on the game and saves a fresh, consistent package snapshot.
 func (r *GameExportsRepo) Start(ctx context.Context, gameID, userID int, id, source string) (*Attempt, error) {
 	tx, err := r.pool.Begin(ctx)
 	if err != nil {
 		return nil, err
 	}
 	defer func() { _ = tx.Rollback(ctx) }()
-	var status string
-	var started, ended *time.Time
+	var lockedID int
 	err = tx.QueryRow(
 		ctx,
 		`SELECT
-			status,
-			started_at,
-			ended_at
+			id
 		FROM games
 		WHERE id = $1
 		FOR UPDATE`,
 		gameID,
-	).Scan(&status, &started, &ended)
+	).Scan(&lockedID)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return nil, ErrGameNotFound
 	}
 	if err != nil {
 		return nil, err
 	}
-	existing, err := readAttempt(ctx, tx, gameID, true, true)
-	if err == nil {
-		return existing, tx.Commit(ctx)
-	}
-	if !errors.Is(err, ErrExportNotFound) {
-		return nil, err
-	}
-	if status != "draft" || started != nil || ended != nil {
-		return nil, ErrNotExportable
-	}
-	var progress bool
-	err = tx.QueryRow(
+	// Retain ready packages for late results; fence unfinished workers before replacing them.
+	_, err = tx.Exec(
 		ctx,
-		`SELECT
-			EXISTS (
-			SELECT
-				1
-			FROM teams t
-			WHERE t.game_id = $1 AND (
-				EXISTS (
-					SELECT
-						1
-					FROM actions a
-					WHERE a.team_id = t.id
-				)
-				OR EXISTS (
-					SELECT
-						1
-					FROM applications a
-					WHERE a.team_id = t.id
-				)
-				OR EXISTS (
-					SELECT
-						1
-					FROM team_answers a
-					WHERE a.team_id = t.id
-				)
-			)
-		)`,
+		`UPDATE game_exports
+		SET released_at = clock_timestamp(),
+			state = CASE WHEN state = 'preparing' THEN 'failed' ELSE state END,
+			failure = CASE WHEN state = 'preparing' THEN 'Подготовка заменена новым пакетом.' ELSE failure END,
+			owner_id = NULL, lease_until = NULL
+		WHERE game_id = $1 AND released_at IS NULL AND state IN ('preparing', 'ready')`,
 		gameID,
-	).Scan(&progress)
+	)
 	if err != nil {
 		return nil, err
 	}
-	if progress {
-		return nil, ErrNotExportable
-	}
 	snapshot, err := captureSnapshot(ctx, tx, gameID)
 	if err != nil {
 		return nil, err
@@ -147,6 +111,75 @@ func (r *GameExportsRepo) Get(ctx context.Context, gameID int) (*Attempt, error)
 	return readAttempt(ctx, r.pool, gameID, false, false)
 }

+// ResetLocalMode releases only the expected latest export, preserving ready archives for late results.
+func (r *GameExportsRepo) ResetLocalMode(ctx context.Context, gameID int, exportID string) error {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	var lockedID int
+	err = tx.QueryRow(
+		ctx,
+		`SELECT
+			id
+		FROM games
+		WHERE id = $1
+		FOR UPDATE`,
+		gameID,
+	).Scan(&lockedID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return ErrGameNotFound
+	}
+	if err != nil {
+		return err
+	}
+	var latestID, state string
+	var releasedAt *time.Time
+	err = tx.QueryRow(
+		ctx,
+		`SELECT
+			id,
+			state,
+			released_at
+		FROM game_exports
+		WHERE game_id = $1
+		ORDER BY created_at DESC, id
+		LIMIT 1
+		FOR UPDATE`,
+		gameID,
+	).Scan(&latestID, &state, &releasedAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return ErrExportNotFound
+	}
+	if err != nil {
+		return err
+	}
+	if latestID != exportID {
+		return ErrExportChanged
+	}
+	if releasedAt != nil {
+		return tx.Commit(ctx)
+	}
+	if state != "ready" && state != "preparing" {
+		return ErrExportChanged
+	}
+	_, err = tx.Exec(
+		ctx,
+		`UPDATE game_exports
+		SET released_at = clock_timestamp(),
+			state = CASE WHEN state = 'preparing' THEN 'failed' ELSE state END,
+			failure = CASE WHEN state = 'preparing' THEN 'Подготовка отменена при сбросе локального режима.' ELSE failure END,
+			owner_id = NULL, lease_until = NULL
+		WHERE id = $1`,
+		exportID,
+	)
+	if err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
 // IRowReader is implemented by both the pool and a transaction.
 type IRowReader interface {
 	QueryRow(context.Context, string, ...any) pgx.Row
@@ -169,7 +202,7 @@ func readAttempt(ctx context.Context, reader IRowReader, gameID int, active, wit
 			object_key,
 			created_at
 		FROM game_exports
-		WHERE game_id = $1 AND (NOT $2 OR state IN ('preparing', 'ready'))
+		WHERE game_id = $1 AND (NOT $2 OR (state IN ('preparing', 'ready') AND released_at IS NULL))
 		ORDER BY created_at DESC, id
 		LIMIT 1`,
 		gameID,
@@ -108,9 +108,9 @@ func TestExportSnapshotAndEditableGame(t *testing.T) {
 			requireNoError(t, mapRepo.Update(ctx, game_map_repo.Item{GameID: f.gameID, Kind: "team", ItemID: f.teamID, X: 10}))
 			_, err = reviews.Moderate(ctx, review.ID, "approved")
 			requireNoError(t, err)
-			same, err := f.exports.Start(ctx, f.gameID, f.userID, "not-a-new-attempt", "test")
+			same, err := f.exports.Get(ctx, f.gameID)
 			requireNoError(t, err)
-			if same.ID != attempt.ID || !bytes.Equal(same.Snapshot, attempt.Snapshot) {
+			if same.ID != attempt.ID {
 				t.Fatal("online edits changed an existing export snapshot")
 			}
 			requireNoError(t, f.scenarios.UpdateScenarioByID(ctx, f.scenarioID, "Edited shared story", "", ""))
@@ -163,12 +163,9 @@ func TestExportConcurrentStartAndTeamWrites(t *testing.T) {
 			if changeErr != nil {
 				t.Fatal(changeErr)
 			}
-			var id string
+			ids := map[string]bool{}
 			for attempt := range attempts {
-				if id != "" && id != attempt.ID {
-					t.Fatal("more than one active export")
-				}
-				id = attempt.ID
+				ids[attempt.ID] = true
 				var snapshot struct {
 					Teams []struct {
 						Name string `json:"name"`
@@ -179,11 +176,26 @@ func TestExportConcurrentStartAndTeamWrites(t *testing.T) {
 					t.Fatalf("unexpected team snapshot: %s", snapshot.Teams[0].Name)
 				}
 			}
+			if len(ids) != 8 {
+				t.Fatalf("expected fresh attempts, got %d", len(ids))
+			}
+			current, err := f.exports.Get(ctx, f.gameID)
+			requireNoError(t, err)
+			claimed, err := f.exports.Claim(ctx, "worker")
+			requireNoError(t, err)
+			if claimed.ID != current.ID {
+				t.Fatal("worker claimed an obsolete attempt")
+			}
+			_, err = f.exports.Claim(ctx, "other-worker")
+			if !errors.Is(err, pgx.ErrNoRows) {
+				t.Fatalf("more than one active export: %v", err)
+			}
+
 		})
 	}
 }

-func TestExportRejectsProgressAndRunningGames(t *testing.T) {
+func TestExportAllowsProgressAndRunningGames(t *testing.T) {
 	for _, tc := range []struct {
 		name   string
 		mutate func(exportFixture) error
@@ -205,9 +217,7 @@ func TestExportRejectsProgressAndRunningGames(t *testing.T) {
 			f := newExportFixture(t)
 			requireNoError(t, tc.mutate(f))
 			_, err := f.exports.Start(context.Background(), f.gameID, f.userID, "attempt", "https://game.example")
-			if !errors.Is(err, ErrNotExportable) {
-				t.Fatalf("export allowed: %v", err)
-			}
+			requireNoError(t, err)
 			requireNoError(t, f.teams.UpdateTeam(context.Background(), f.teamID, "Still writable"))
 		})
 	}
@@ -0,0 +1,150 @@
+package game_exports_repo
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+)
+
+func TestResetLocalModePreservesReadyPackageAndGame(t *testing.T) {
+	for _, name := range []string{"ResetLocalModePreservesReadyPackageAndGame"} {
+		t.Run(name, func(t *testing.T) {
+			f := newExportFixture(t)
+			ctx := context.Background()
+			attempt := f.start(t, "ready-attempt")
+			_, err := f.exports.Claim(ctx, "worker")
+			requireNoError(t, err)
+			requireNoError(t, f.exports.Complete(ctx, attempt.ID, "worker", "package", "digest", 123))
+			before, err := f.games.GetGameByID(ctx, f.gameID)
+			requireNoError(t, err)
+			if before.Export == nil {
+				t.Fatal("missing local mode before reset")
+			}
+			requireNoError(t, f.exports.ResetLocalMode(ctx, f.gameID, attempt.ID))
+			requireNoError(t, f.exports.ResetLocalMode(ctx, f.gameID, attempt.ID))
+			after, err := f.games.GetGameByID(ctx, f.gameID)
+			requireNoError(t, err)
+			if after.Export != nil || after.Status != before.Status || after.Name != before.Name {
+				t.Fatal("reset retained local mode or changed the game")
+			}
+			games, err := f.games.GetGames(ctx)
+			requireNoError(t, err)
+			for _, game := range games {
+				if game.ID == f.gameID && game.Export != nil {
+					t.Fatal("game list retained local mode")
+				}
+			}
+			saved, err := f.exports.Get(ctx, f.gameID)
+			requireNoError(t, err)
+			if saved.ID != attempt.ID || saved.State != "ready" || saved.PackageID != "package" {
+				t.Fatal("reset damaged the package needed for late results")
+			}
+			next := f.start(t, "new-attempt")
+			if err := f.exports.ResetLocalMode(ctx, f.gameID, attempt.ID); !errors.Is(err, ErrExportChanged) {
+				t.Fatalf("stale confirmation accepted: %v", err)
+			}
+			current, err := f.exports.Get(ctx, f.gameID)
+			requireNoError(t, err)
+			if current.ID != next.ID || current.State != "preparing" {
+				t.Fatal("stale confirmation changed new export")
+			}
+		})
+	}
+}
+
+func TestResetAndReplacementFencePreparingWorker(t *testing.T) {
+	for _, replace := range []bool{false, true} {
+		t.Run(map[bool]string{false: "reset", true: "replace"}[replace], func(t *testing.T) {
+			f := newExportFixture(t)
+			ctx := context.Background()
+			attempt := f.start(t, "old")
+			_, err := f.exports.Claim(ctx, "worker")
+			requireNoError(t, err)
+			if replace {
+				f.start(t, "new")
+			} else {
+				requireNoError(t, f.exports.ResetLocalMode(ctx, f.gameID, attempt.ID))
+			}
+			if err := f.exports.Complete(ctx, attempt.ID, "worker", "package", "digest", 123); !errors.Is(err, ErrOwnershipLost) {
+				t.Fatalf("cancelled worker published: %v", err)
+			}
+			if err := f.exports.Heartbeat(ctx, attempt.ID, "worker"); !errors.Is(err, ErrOwnershipLost) {
+				t.Fatalf("cancelled worker renewed lease: %v", err)
+			}
+			keys, err := f.exports.CleanupKeys(ctx)
+			requireNoError(t, err)
+			if len(keys) != 1 || keys[0] != attempt.ObjectKey {
+				t.Fatalf("cancelled object cleanup lost: %v", keys)
+			}
+		})
+	}
+}
+
+func TestResetConcurrentReplacement(t *testing.T) {
+	for _, name := range []string{"ResetConcurrentReplacement"} {
+		t.Run(name, func(t *testing.T) {
+			f := newExportFixture(t)
+			ctx := context.Background()
+			attempt := f.start(t, "old")
+			var wg sync.WaitGroup
+			var resetErr, startErr error
+			wg.Add(2)
+			go func() { defer wg.Done(); resetErr = f.exports.ResetLocalMode(ctx, f.gameID, attempt.ID) }()
+			go func() { defer wg.Done(); _, startErr = f.exports.Start(ctx, f.gameID, f.userID, "new", "source") }()
+			wg.Wait()
+			requireNoError(t, startErr)
+			if resetErr != nil && !errors.Is(resetErr, ErrExportChanged) {
+				t.Fatal(resetErr)
+			}
+			game, err := f.games.GetGameByID(ctx, f.gameID)
+			requireNoError(t, err)
+			if game.Export == nil || game.Export.ID != "new" {
+				t.Fatal("concurrent reset released the new export")
+			}
+		})
+	}
+}
+
+func TestResetMissingGameOrExport(t *testing.T) {
+	for _, name := range []string{"ResetMissingGameOrExport"} {
+		t.Run(name, func(t *testing.T) {
+			f := newExportFixture(t)
+			ctx := context.Background()
+			if err := f.exports.ResetLocalMode(ctx, f.gameID+1, "missing"); !errors.Is(err, ErrGameNotFound) {
+				t.Fatal(err)
+			}
+			if err := f.exports.ResetLocalMode(ctx, f.gameID, "missing"); !errors.Is(err, ErrExportNotFound) {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestReleaseMigrationRollback(t *testing.T) {
+	for _, name := range []string{"ReleaseMigrationRollback"} {
+		t.Run(name, func(t *testing.T) {
+			f := newExportFixture(t)
+			ctx := context.Background()
+			requireNoError(t, f.releaseMigration(t, true))
+			requireNoError(t, f.releaseMigration(t, false))
+			attempt := f.start(t, "first")
+			_, err := f.exports.Claim(ctx, "worker")
+			requireNoError(t, err)
+			requireNoError(t, f.exports.Complete(ctx, attempt.ID, "worker", "package-1", "digest", 1))
+			second := f.start(t, "second")
+			_, err = f.exports.Claim(ctx, "worker")
+			requireNoError(t, err)
+			requireNoError(t, f.exports.Complete(ctx, second.ID, "worker", "package-2", "digest", 1))
+			// Rollback must refuse incompatible history without removing the current export or schema.
+			if err := f.releaseMigration(t, true); err == nil {
+				t.Fatal("rollback allowed incompatible ready history")
+			}
+			current, err := f.exports.Get(ctx, f.gameID)
+			requireNoError(t, err)
+			if current.ID != second.ID || current.State != "ready" {
+				t.Fatal("rollback damaged export history")
+			}
+		})
+	}
+}
@@ -78,7 +78,7 @@ func (r *GamesRepo) GetGameByID(
 			games.started_at,
 			games.ended_at,
 			(SELECT
-				CASE WHEN EXISTS (
+				CASE WHEN e.released_at IS NOT NULL OR EXISTS (
 					SELECT 1
 					FROM game_returns r
 					WHERE r.export_id = e.id
@@ -152,7 +152,7 @@ func (r *GamesRepo) GetGames(
 			games.format,
 			games.status,
 			(SELECT
-				CASE WHEN EXISTS (
+				CASE WHEN e.released_at IS NOT NULL OR EXISTS (
 					SELECT 1
 					FROM game_returns r
 					WHERE r.export_id = e.id