feat: backend executor implement

2026-08-16 17:12:06 +08:00
parent 390e0565d3
commit 7755da72e7
31 changed files with 2762 additions and 184 deletions
@@ -0,0 +1,185 @@
+package transaction
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+)
+
+const backendService = "backend"
+
+// BackendContainerDeployment is the last container slot committed for backend.
+// Runtime inspection remains authoritative for the current external state; this
+// record distinguishes a fresh installation from loss or drift after deployment.
+type BackendContainerDeployment struct {
+	ActivePort    int
+	ContainerName string
+	ImageDigest   string
+	ContainerID   string
+	TransactionID string
+	UpdatedAt     time.Time
+}
+
+// BackendContainerDeployment returns the single committed backend container state.
+func (s *Store) BackendContainerDeployment(ctx context.Context) (BackendContainerDeployment, error) {
+	return scanBackendContainerDeployment(s.db.QueryRowContext(ctx, `
+        SELECT active_port, container_name, image_digest, container_id,
+               transaction_id, updated_at
+          FROM backend_container_deployment
+         WHERE singleton_id = 1`))
+}
+
+// HasCommittedBackendContainerTransactionHistory reports whether this store
+// contains a committed container backend transaction. This is the legacy
+// deployment evidence used when a database predates the deployment table.
+// Failed and rolled-back first-install attempts do not mark a machine deployed.
+func (s *Store) HasCommittedBackendContainerTransactionHistory(ctx context.Context) (bool, error) {
+	var found int
+	if err := s.db.QueryRowContext(ctx, `
+        SELECT EXISTS (
+            SELECT 1
+              FROM transactions
+             WHERE service = ?
+               AND idempotency_key GLOB 'backend:container:*'
+               AND state = ?
+        )`, backendService, StateCommitted).Scan(&found); err != nil {
+		return false, fmt.Errorf("query committed backend container transaction history: %w", err)
+	}
+	return found == 1, nil
+}
+
+// CommitBackendContainerDeployment atomically records the active container and
+// commits its transaction. A crash cannot leave COMMITTED without the matching
+// deployment row, or publish a deployment row for an unfinished transaction.
+func (s *Store) CommitBackendContainerDeployment(
+	ctx context.Context,
+	transactionID string,
+	deployment BackendContainerDeployment,
+	message string,
+) (Transaction, error) {
+	if transactionID == "" || strings.TrimSpace(transactionID) != transactionID {
+		return Transaction{}, errors.New("exact transaction ID is required")
+	}
+	if err := validateBackendContainerDeployment(deployment); err != nil {
+		return Transaction{}, err
+	}
+
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
+	if err != nil {
+		return Transaction{}, fmt.Errorf("begin backend container deployment commit: %w", err)
+	}
+	defer tx.Rollback()
+
+	record, err := getTransactionByID(ctx, tx, transactionID)
+	if err != nil {
+		return Transaction{}, err
+	}
+	if record.Service != backendService {
+		return Transaction{}, fmt.Errorf("transaction %s service must be %s", transactionID, backendService)
+	}
+	if record.State != StateDraining {
+		return Transaction{}, &TransitionError{From: record.State, To: StateCommitted}
+	}
+
+	now := s.now().UTC()
+	_, err = tx.ExecContext(ctx, `
+        INSERT INTO backend_container_deployment (
+            singleton_id, active_port, container_name, image_digest,
+            container_id, transaction_id, updated_at
+        ) VALUES (1, ?, ?, ?, ?, ?, ?)
+        ON CONFLICT(singleton_id) DO UPDATE SET
+            active_port = excluded.active_port,
+            container_name = excluded.container_name,
+            image_digest = excluded.image_digest,
+            container_id = excluded.container_id,
+            transaction_id = excluded.transaction_id,
+            updated_at = excluded.updated_at`,
+		deployment.ActivePort,
+		deployment.ContainerName,
+		deployment.ImageDigest,
+		deployment.ContainerID,
+		transactionID,
+		formatTime(now),
+	)
+	if err != nil {
+		return Transaction{}, fmt.Errorf("write backend container deployment: %w", err)
+	}
+
+	result, err := tx.ExecContext(ctx, `
+        UPDATE transactions
+           SET state = ?, version = version + 1, updated_at = ?
+         WHERE id = ? AND version = ? AND state = ?`,
+		StateCommitted,
+		formatTime(now),
+		transactionID,
+		record.Version,
+		StateDraining,
+	)
+	if err != nil {
+		return Transaction{}, fmt.Errorf("commit backend container transaction state: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return Transaction{}, fmt.Errorf("read committed backend container transaction rows: %w", err)
+	}
+	if rows != 1 {
+		return Transaction{}, errors.New("backend container transaction changed concurrently")
+	}
+	if err := insertEvent(ctx, tx, transactionID, "", "TRANSACTION_STATE_CHANGED", record.State, StateCommitted, message, now); err != nil {
+		return Transaction{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return Transaction{}, fmt.Errorf("commit backend container deployment: %w", err)
+	}
+
+	record.State = StateCommitted
+	record.Version++
+	record.UpdatedAt = now
+	return record, nil
+}
+
+func validateBackendContainerDeployment(deployment BackendContainerDeployment) error {
+	if deployment.ActivePort != 8080 && deployment.ActivePort != 8081 {
+		return fmt.Errorf("backend container deployment port must be 8080 or 8081: %d", deployment.ActivePort)
+	}
+	for _, field := range []struct {
+		name  string
+		value string
+	}{
+		{"container name", deployment.ContainerName},
+		{"image digest", deployment.ImageDigest},
+		{"container ID", deployment.ContainerID},
+	} {
+		if field.value == "" || strings.TrimSpace(field.value) != field.value {
+			return fmt.Errorf("exact backend container %s is required", field.name)
+		}
+	}
+	return nil
+}
+
+func scanBackendContainerDeployment(row rowScanner) (BackendContainerDeployment, error) {
+	var deployment BackendContainerDeployment
+	var updatedAt string
+	if err := row.Scan(
+		&deployment.ActivePort,
+		&deployment.ContainerName,
+		&deployment.ImageDigest,
+		&deployment.ContainerID,
+		&deployment.TransactionID,
+		&updatedAt,
+	); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return BackendContainerDeployment{}, ErrNotFound
+		}
+		return BackendContainerDeployment{}, fmt.Errorf("scan backend container deployment: %w", err)
+	}
+	parsed, err := parseTime(updatedAt)
+	if err != nil {
+		return BackendContainerDeployment{}, fmt.Errorf("parse backend container deployment update time: %w", err)
+	}
+	deployment.UpdatedAt = parsed
+	return deployment, nil
+}
@@ -93,7 +93,28 @@ func (c *Coordinator) ExecuteStep(ctx context.Context, transactionID string, int
 		return step, nil
 	}
 	if step.Status == StepFailed {
-		return step, fmt.Errorf("external step already failed: %s", step.Error)
+		inspection, inspectErr := operation.Inspect(ctx)
+		if inspectErr != nil {
+			return step, c.uncertain(ctx, transactionID, intent.Key, fmt.Errorf("inspect failed step: %w", inspectErr))
+		}
+		switch inspection.Status {
+		case InspectionApplied:
+			if _, err := c.store.ReopenFailedStep(ctx, transactionID, intent.Key, "manual resume found failed step applied"); err != nil {
+				return step, err
+			}
+			return c.completeApplied(ctx, transactionID, intent.Key, inspection)
+		case InspectionNotApplied:
+			reopened, err := c.store.ReopenFailedStep(ctx, transactionID, intent.Key, "manual resume found failed step not applied")
+			if err != nil {
+				return step, err
+			}
+			step = reopened
+			created = true
+		case InspectionUnknown:
+			return step, c.uncertain(ctx, transactionID, intent.Key, errors.New("inspect failed step returned UNKNOWN"))
+		default:
+			return step, c.uncertain(ctx, transactionID, intent.Key, fmt.Errorf("inspect failed step returned invalid status %q", inspection.Status))
+		}
 	}

 	if !created {
@@ -72,6 +72,56 @@ func TestCoordinatorLeavesIntentPendingWhenInspectionIsUnknown(t *testing.T) {
 	}
 }

+func TestCoordinatorRetriesFailedStepOnlyAfterManualResumeInspection(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	store := openTestStore(t)
+	record := createTestTransaction(t, store, "retry-failed")
+	intent := StepIntent{Key: "restore-gateway", Name: "restore gateway", Intent: json.RawMessage(`{}`)}
+	operation := &fakeOperation{applyErr: errors.New("gateway executable unavailable"), result: json.RawMessage(`{"restored":true}`)}
+	coordinator := newTestCoordinator(t, store)
+
+	failed, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
+	if err == nil || failed.Status != StepFailed {
+		t.Fatalf("unexpected initial failed step: step=%+v err=%v", failed, err)
+	}
+	if operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 1 {
+		t.Fatalf("initial call retried unexpectedly: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load())
+	}
+
+	operation.applyErr = nil
+	recovered, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
+	if err != nil || recovered.Status != StepSucceeded {
+		t.Fatalf("recover failed step: step=%+v err=%v", recovered, err)
+	}
+	if operation.applyCalls.Load() != 2 || operation.inspectCalls.Load() != 3 {
+		t.Fatalf("unexpected manual recovery calls: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load())
+	}
+}
+
+func TestCoordinatorCompletesFailedStepAlreadyAppliedBeforeManualResume(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	store := openTestStore(t)
+	record := createTestTransaction(t, store, "recover-failed-applied")
+	intent := StepIntent{Key: "restore-gateway", Name: "restore gateway", Intent: json.RawMessage(`{}`)}
+	operation := &fakeOperation{applyErr: errors.New("gateway reload result unavailable"), result: json.RawMessage(`{"restored":true}`)}
+	coordinator := newTestCoordinator(t, store)
+
+	if _, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation); err == nil {
+		t.Fatal("expected initial external step failure")
+	}
+	operation.applyErr = nil
+	operation.applied = true
+	recovered, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
+	if err != nil || recovered.Status != StepSucceeded {
+		t.Fatalf("complete externally applied failed step: step=%+v err=%v", recovered, err)
+	}
+	if operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 2 {
+		t.Fatalf("externally applied step was repeated: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load())
+	}
+}
+
 func TestCoordinatorExclusiveExecutionHonorsContext(t *testing.T) {
 	t.Parallel()
 	store := openTestStore(t)
@@ -17,7 +17,7 @@ import (
 	"github.com/ncruces/go-sqlite3/driver"
 )

-const schemaVersion = 1
+const schemaVersion = 2

 const schemaV1 = `
 CREATE TABLE transactions (
@@ -70,13 +70,27 @@ CREATE INDEX transaction_events_by_transaction
 PRAGMA user_version = 1;
 `

+const schemaV2 = `
+CREATE TABLE backend_container_deployment (
+    singleton_id   INTEGER PRIMARY KEY CHECK (singleton_id = 1),
+    active_port    INTEGER NOT NULL CHECK (active_port IN (8080, 8081)),
+    container_name TEXT NOT NULL,
+    image_digest   TEXT NOT NULL,
+    container_id   TEXT NOT NULL,
+    transaction_id TEXT NOT NULL REFERENCES transactions(id),
+    updated_at     TEXT NOT NULL
+) STRICT;
+
+PRAGMA user_version = 2;
+`
+
 // Store 是服务端 SQLite 事务记录。一个进程只应创建一个 Store。
 type Store struct {
 	db  *sql.DB
 	now func() time.Time
 }

-// OpenStore 打开本地 SQLite，并强制校验第一版持久化参数。
+// OpenStore 打开本地 SQLite，并强制校验持久化参数和 schema 版本。
 func OpenStore(ctx context.Context, path string) (*Store, error) {
 	if path == "" {
 		return nil, errors.New("sqlite path is required")
@@ -168,8 +182,21 @@ func migrate(ctx context.Context, db *sql.DB) error {
 		return fmt.Errorf("begin sqlite migration: %w", err)
 	}
 	defer tx.Rollback()
-	if _, err := tx.ExecContext(ctx, schemaV1); err != nil {
-		return fmt.Errorf("apply sqlite schema version 1: %w", err)
+	for version < schemaVersion {
+		nextVersion := version + 1
+		var script string
+		switch nextVersion {
+		case 1:
+			script = schemaV1
+		case 2:
+			script = schemaV2
+		default:
+			return fmt.Errorf("sqlite migration script is missing for version %d", nextVersion)
+		}
+		if _, err := tx.ExecContext(ctx, script); err != nil {
+			return fmt.Errorf("apply sqlite schema version %d: %w", nextVersion, err)
+		}
+		version = nextVersion
 	}
 	if err := tx.Commit(); err != nil {
 		return fmt.Errorf("commit sqlite migration: %w", err)
@@ -199,7 +226,37 @@ func (s *Store) CreateTransaction(ctx context.Context, request CreateRequest) (r

 	existing, err := getTransactionByIdempotencyKey(ctx, tx, request.IdempotencyKey)
 	if err == nil {
-		return existing, false, nil
+		if existing.State != StateFailed && existing.State != StateRolledBack {
+			return existing, false, nil
+		}
+		archivedKey := existing.IdempotencyKey + ":terminal:" + existing.ID
+		result, updateErr := tx.ExecContext(ctx, `
+            UPDATE transactions
+               SET idempotency_key = ?, version = version + 1, updated_at = ?
+             WHERE id = ? AND version = ? AND idempotency_key = ?
+               AND state IN (?, ?)`,
+			archivedKey,
+			formatTime(now),
+			existing.ID,
+			existing.Version,
+			existing.IdempotencyKey,
+			StateFailed,
+			StateRolledBack,
+		)
+		if updateErr != nil {
+			return Transaction{}, false, fmt.Errorf("archive terminal transaction idempotency key: %w", updateErr)
+		}
+		rows, rowsErr := result.RowsAffected()
+		if rowsErr != nil {
+			return Transaction{}, false, fmt.Errorf("read archived transaction rows: %w", rowsErr)
+		}
+		if rows != 1 {
+			return Transaction{}, false, errors.New("terminal transaction changed concurrently")
+		}
+		if eventErr := insertEvent(ctx, tx, existing.ID, "", "TRANSACTION_RETRY_RELEASED", existing.State, existing.State, "terminal transaction idempotency key archived for manual retry", now); eventErr != nil {
+			return Transaction{}, false, eventErr
+		}
+		err = ErrNotFound
 	}
 	if !errors.Is(err, ErrNotFound) {
 		return Transaction{}, false, err
@@ -469,6 +526,59 @@ func (s *Store) CompleteStep(ctx context.Context, transactionID, stepKey string,
 	return record, nil
 }

+// ReopenFailedStep makes one manually resumed external step pending again.
+// The coordinator calls this only after Inspect has established an exact
+// APPLIED or NOT_APPLIED state; UNKNOWN never reopens a failed step.
+func (s *Store) ReopenFailedStep(ctx context.Context, transactionID, stepKey, message string) (Step, error) {
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
+	if err != nil {
+		return Step{}, fmt.Errorf("begin reopen failed step: %w", err)
+	}
+	defer tx.Rollback()
+	record, err := getStep(ctx, tx, transactionID, stepKey)
+	if err != nil {
+		return Step{}, err
+	}
+	if record.Status == StepIntentRecorded {
+		return record, nil
+	}
+	if record.Status != StepFailed {
+		return Step{}, ErrStepNotPending
+	}
+	now := s.now().UTC()
+	result, err := tx.ExecContext(ctx, `
+        UPDATE transaction_steps
+           SET status = ?, result_json = NULL, error_message = '', updated_at = ?
+         WHERE transaction_id = ? AND step_key = ? AND status = ?`,
+		StepIntentRecorded,
+		formatTime(now),
+		transactionID,
+		stepKey,
+		StepFailed,
+	)
+	if err != nil {
+		return Step{}, fmt.Errorf("reopen failed step: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return Step{}, fmt.Errorf("read reopened step rows: %w", err)
+	}
+	if rows != 1 {
+		return Step{}, errors.New("failed step changed concurrently")
+	}
+	if err := insertEvent(ctx, tx, transactionID, stepKey, "STEP_REOPENED", "", "", message, now); err != nil {
+		return Step{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return Step{}, fmt.Errorf("commit reopened step: %w", err)
+	}
+	record.Status = StepIntentRecorded
+	record.Result = nil
+	record.Error = ""
+	record.UpdatedAt = now
+	return record, nil
+}
+
 // PendingSteps 返回重启后必须先核对实际外部状态的步骤。
 func (s *Store) PendingSteps(ctx context.Context, transactionID string) ([]Step, error) {
 	rows, err := s.db.QueryContext(ctx, `
@@ -4,12 +4,135 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"net/url"
 	"path/filepath"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/ncruces/go-sqlite3/driver"
 )

+func TestStoreMigratesVersionOneAndCommitsBackendContainerDeployment(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	databasePath := filepath.Join(t.TempDir(), "transaction.db")
+	dsn := (&url.URL{Scheme: "file", Path: databasePath}).String()
+	database, err := driver.Open(dsn)
+	if err != nil {
+		t.Fatalf("open version one database: %v", err)
+	}
+	if _, err := database.ExecContext(ctx, schemaV1); err != nil {
+		_ = database.Close()
+		t.Fatalf("create version one database: %v", err)
+	}
+	if err := database.Close(); err != nil {
+		t.Fatalf("close version one database: %v", err)
+	}
+
+	store, err := OpenStore(ctx, databasePath)
+	if err != nil {
+		t.Fatalf("migrate transaction store: %v", err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+
+	if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("unexpected deployment before commit: %v", err)
+	}
+	hasHistory, err := store.HasCommittedBackendContainerTransactionHistory(ctx)
+	if err != nil || hasHistory {
+		t.Fatalf("unexpected empty backend history: found=%t err=%v", hasHistory, err)
+	}
+	nativeRecord, _, err := store.CreateTransaction(ctx, CreateRequest{
+		ID:             "native-backend-transaction",
+		IdempotencyKey: "backend:native-request",
+		Source:         "test",
+		Service:        "backend",
+	})
+	if err != nil {
+		t.Fatalf("create native backend transaction: %v", err)
+	}
+	if _, err := store.Transition(ctx, nativeRecord.ID, StateFailed, "finish native backend transaction"); err != nil {
+		t.Fatalf("finish native backend transaction: %v", err)
+	}
+	hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx)
+	if err != nil || hasHistory {
+		t.Fatalf("native backend history was treated as container history: found=%t err=%v", hasHistory, err)
+	}
+
+	record, created, err := store.CreateTransaction(ctx, CreateRequest{
+		ID:             "container-deployment-transaction",
+		IdempotencyKey: "backend:container:deployment-request",
+		Source:         "test",
+		Service:        "backend",
+	})
+	if err != nil || !created {
+		t.Fatalf("create backend container transaction: record=%+v created=%t err=%v", record, created, err)
+	}
+	for _, state := range []State{StateValidating, StatePrepared, StateStarting, StateSwitching, StateVerifying, StateDraining} {
+		if _, err := store.Transition(ctx, record.ID, state, "test transition"); err != nil {
+			t.Fatalf("transition backend container transaction to %s: %v", state, err)
+		}
+	}
+	hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx)
+	if err != nil || hasHistory {
+		t.Fatalf("unfinished backend container transaction was treated as committed history: found=%t err=%v", hasHistory, err)
+	}
+
+	committed, err := store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{
+		ActivePort:    8081,
+		ContainerName: "backend-8081",
+		ImageDigest:   "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+		ContainerID:   "container-backend-8081",
+	}, "container deployment committed")
+	if err != nil || committed.State != StateCommitted {
+		t.Fatalf("commit backend container deployment: record=%+v err=%v", committed, err)
+	}
+	hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx)
+	if err != nil || !hasHistory {
+		t.Fatalf("committed backend container history was not recorded: found=%t err=%v", hasHistory, err)
+	}
+	deployment, err := store.BackendContainerDeployment(ctx)
+	if err != nil {
+		t.Fatalf("read backend container deployment: %v", err)
+	}
+	if deployment.ActivePort != 8081 || deployment.ContainerName != "backend-8081" || deployment.ContainerID != "container-backend-8081" || deployment.TransactionID != record.ID {
+		t.Fatalf("unexpected backend container deployment: %+v", deployment)
+	}
+	var version int
+	if err := store.db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil || version != schemaVersion {
+		t.Fatalf("unexpected migrated schema version: version=%d err=%v", version, err)
+	}
+}
+
+func TestBackendContainerDeploymentCommitRequiresDrainingTransaction(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	store := openTestStore(t)
+	record, _, err := store.CreateTransaction(ctx, CreateRequest{
+		ID:             "container-deployment-wrong-state",
+		IdempotencyKey: "container-deployment-wrong-state-request",
+		Source:         "test",
+		Service:        "backend",
+	})
+	if err != nil {
+		t.Fatalf("create backend transaction: %v", err)
+	}
+	_, err = store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{
+		ActivePort:    8080,
+		ContainerName: "backend-8080",
+		ImageDigest:   "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+		ContainerID:   "container-backend-8080",
+	}, "must fail")
+	var transitionErr *TransitionError
+	if !errors.As(err, &transitionErr) {
+		t.Fatalf("expected deployment commit transition error, got %v", err)
+	}
+	if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("failed commit wrote backend deployment: %v", err)
+	}
+}
+
 func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -51,6 +174,26 @@ func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T)
 	if _, err := store.Transition(ctx, created.ID, StateFailed, "test terminal state"); err != nil {
 		t.Fatalf("finish first transaction: %v", err)
 	}
+	retriedAfterFailure, isNew, err := store.CreateTransaction(ctx, CreateRequest{
+		ID:             "transaction-1-retry",
+		IdempotencyKey: request.IdempotencyKey,
+		Source:         request.Source,
+		Service:        request.Service,
+		Request:        request.Request,
+	})
+	if err != nil || !isNew || retriedAfterFailure.ID != "transaction-1-retry" {
+		t.Fatalf("retry failed transaction: record=%+v new=%v err=%v", retriedAfterFailure, isNew, err)
+	}
+	archived, err := store.Transaction(ctx, created.ID)
+	if err != nil {
+		t.Fatalf("read archived failed transaction: %v", err)
+	}
+	if archived.IdempotencyKey != request.IdempotencyKey+":terminal:"+created.ID {
+		t.Fatalf("unexpected archived idempotency key: %q", archived.IdempotencyKey)
+	}
+	if _, err := store.Transition(ctx, retriedAfterFailure.ID, StateFailed, "finish retried transaction"); err != nil {
+		t.Fatalf("finish retried transaction: %v", err)
+	}
 	second, isNew, err := store.CreateTransaction(ctx, CreateRequest{
 		ID:             "transaction-2",
 		IdempotencyKey: "request-2",