diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_send_now_fallback_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_send_now_fallback_test.go new file mode 100644 index 0000000000..69ef78ccbc --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/repository_send_now_fallback_test.go @@ -0,0 +1,185 @@ +package messagequeue + +import ( + "context" + "errors" + "fmt" + "testing" +) + +func TestSQLiteLegacyTransferRejectsIdentityAwareSendNowClaim(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t).(*sqliteRepository) + sourceIdentity := QueueSessionIdentity{ + TaskID: "task-1", + SessionID: "session-old", + SessionIncarnationID: "incarnation-old", + } + destinationIdentity := QueueSessionIdentity{ + TaskID: sourceIdentity.TaskID, + SessionID: "session-new", + SessionIncarnationID: "incarnation-new", + } + seedQueueSessionIdentity(t, repo, sourceIdentity) + seedQueueSessionIdentity(t, repo, destinationIdentity) + source := &QueuedMessage{ + ID: "source", + TaskID: sourceIdentity.TaskID, + SessionID: sourceIdentity.SessionID, + Content: "identity-aware source", + QueuedBy: QueuedByUser, + } + if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + if _, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source}); err != nil { + t.Fatal(err) + } + + err := repo.TransferSession(ctx, sourceIdentity.SessionID, destinationIdentity.SessionID) + if !errors.Is(err, ErrSessionIdentityMismatch) { + t.Fatalf("legacy transfer error = %v, want ErrSessionIdentityMismatch", err) + } + pending, err := repo.ListPendingSendNowClaims(ctx) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 || pending[0].Claim.Identity != sourceIdentity { + t.Fatalf("pending claims after rejected transfer = %#v, want source identity", pending) + } +} + +func TestDurableTransferRejectsIdentityAwareClaimWhenDestinationIdentityIsUnavailable(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t).(*sqliteRepository) + service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession) + sourceIdentity := QueueSessionIdentity{ + TaskID: "task-1", + SessionID: "session-old", + SessionIncarnationID: "incarnation-old", + } + destinationIdentity := QueueSessionIdentity{TaskID: sourceIdentity.TaskID, SessionID: "session-new"} + seedQueueSessionIdentity(t, repo, sourceIdentity) + seedQueueSessionIdentity(t, repo, destinationIdentity) + source := &QueuedMessage{ + ID: "source", + TaskID: sourceIdentity.TaskID, + SessionID: sourceIdentity.SessionID, + Content: "identity-aware source", + QueuedBy: QueuedByUser, + } + if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + if _, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source}); err != nil { + t.Fatal(err) + } + + err := service.TransferSessionWithDurableAttachmentPreparation( + ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil, + ) + if !errors.Is(err, ErrSessionIdentityMismatch) { + t.Fatalf("durable transfer error = %v, want ErrSessionIdentityMismatch", err) + } + pending, err := repo.ListPendingSendNowClaims(ctx) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 || pending[0].Claim.Identity != sourceIdentity { + t.Fatalf("pending claims after rejected durable transfer = %#v, want source identity", pending) + } +} + +func TestTransfersRejectReincarnatedIdentityAwareSendNowClaim(t *testing.T) { + transfers := []struct { + name string + run func(context.Context, *sqliteRepository, QueueSessionIdentity, QueueSessionIdentity) error + }{ + { + name: "direct", + run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error { + return repo.TransferSessionIdentities(ctx, source, destination) + }, + }, + { + name: "durable", + run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error { + service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession) + return service.TransferSessionWithDurableAttachmentPreparation( + ctx, source.TaskID, source.SessionID, destination.SessionID, nil, nil, + ) + }, + }, + } + + for _, transfer := range transfers { + for _, accepted := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/accepted=%t", transfer.name, accepted), func(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t).(*sqliteRepository) + originalSource := QueueSessionIdentity{ + TaskID: "task-1", SessionID: "session-old", SessionIncarnationID: "incarnation-old", + } + liveSource := originalSource + liveSource.SessionIncarnationID = "incarnation-recreated" + destination := QueueSessionIdentity{ + TaskID: originalSource.TaskID, SessionID: "session-new", SessionIncarnationID: "incarnation-new", + } + seedQueueSessionIdentity(t, repo, originalSource) + seedQueueSessionIdentity(t, repo, destination) + first := &QueuedMessage{ + ID: "first", TaskID: originalSource.TaskID, SessionID: originalSource.SessionID, + Content: "first", QueuedBy: QueuedByUser, + } + second := &QueuedMessage{ + ID: "second", TaskID: originalSource.TaskID, SessionID: originalSource.SessionID, + Content: "second", QueuedBy: QueuedByUser, + } + if err := repo.InsertForSession(ctx, originalSource, first, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + if err := repo.InsertForSession(ctx, originalSource, second, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + claim, err := repo.ClaimSendNowForSession(ctx, originalSource, []QueuedMessage{*first, *second}) + if err != nil { + t.Fatal(err) + } + if accepted { + if err := repo.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil { + t.Fatal(err) + } + } + seedQueueSessionIdentity(t, repo, liveSource) + + err = transfer.run(ctx, repo, liveSource, destination) + if !errors.Is(err, ErrSessionIdentityMismatch) { + t.Fatalf("transfer error = %v, want ErrSessionIdentityMismatch", err) + } + pending, err := repo.ListPendingSendNowClaims(ctx) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 { + t.Fatalf("pending claims = %#v, want one preserved claim", pending) + } + preserved := pending[0] + if preserved.Claim.ClaimID != claim.ClaimID || preserved.Claim.Identity != originalSource || + preserved.Accepted != accepted || preserved.Claim.Dispatch.Content != "first\n\nsecond" { + t.Fatalf("preserved claim = %#v", preserved) + } + contents := []string{preserved.Claim.Sources[0].Content, preserved.Claim.Sources[1].Content} + if contents[0] != "first" || contents[1] != "second" { + t.Fatalf("preserved claim sources = %#v", preserved.Claim.Sources) + } + destinationEntries, err := repo.ListBySession(ctx, destination.SessionID) + if err != nil { + t.Fatal(err) + } + if len(destinationEntries) != 0 { + t.Fatalf("destination entries after rejected transfer = %#v", destinationEntries) + } + }) + } + } +} diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery.go b/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery.go index e734501d96..c32fc52ab4 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery.go @@ -140,6 +140,9 @@ func (r *sqliteRepository) transferPendingSendNowClaimTx( ctx context.Context, tx *sqlx.Tx, oldSessionID, newSessionID string, + source, destination *QueueSessionIdentity, + sourceGeneration int64, + queuePositionOffset int64, ) error { var claimID, claimJSON string err := tx.QueryRowxContext(ctx, r.db.Rebind(` @@ -174,14 +177,30 @@ func (r *sqliteRepository) transferPendingSendNowClaimTx( if err != nil || sessionID != oldSessionID || claim.Dispatch.SessionID != oldSessionID { return ErrSendNowClaimChanged } + if claim.Identity.SessionIncarnationID != "" { + if source == nil || claim.Identity != *source { + return ErrSessionIdentityMismatch + } + if claim.OperationGeneration != sourceGeneration || claim.SessionGeneration != sourceGeneration { + return ErrSendNowClaimChanged + } + } + if destination == nil && claim.Identity.SessionIncarnationID != "" { + return ErrSessionIdentityMismatch + } for sourceIndex := range claim.Sources { claim.Sources[sourceIndex].SessionID = newSessionID + claim.Sources[sourceIndex].Position += queuePositionOffset } claim.Dispatch.SessionID = newSessionID + if destination != nil { + claim.Identity = *destination + } claim.SessionGeneration, err = r.getSendNowGenerationTx(ctx, tx, newSessionID) if err != nil { return err } + claim.OperationGeneration = claim.SessionGeneration claimJSONBytes, err := json.Marshal(&claim) if err != nil { return fmt.Errorf("marshal transferred Send Now claim: %w", err) diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery_test.go index c369a8bd82..da550d026f 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery_test.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery_test.go @@ -113,6 +113,201 @@ func TestSQLiteTransferSessionMovesPendingSendNowClaim(t *testing.T) { } } +func TestSQLiteTransferSessionIdentitiesRebindsPendingSendNowClaim(t *testing.T) { + for _, accepted := range []bool{false, true} { + t.Run(fmt.Sprintf("accepted=%t", accepted), func(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t).(*sqliteRepository) + sourceIdentity := QueueSessionIdentity{ + TaskID: "task-1", + SessionID: "session-old", + SessionIncarnationID: "incarnation-old", + } + destinationIdentity := QueueSessionIdentity{ + TaskID: "task-1", + SessionID: "session-new", + SessionIncarnationID: "incarnation-new", + } + seedQueueSessionIdentity(t, repo, sourceIdentity) + seedQueueSessionIdentity(t, repo, destinationIdentity) + destinationPrime := &QueuedMessage{ + ID: "destination-prime", TaskID: destinationIdentity.TaskID, SessionID: destinationIdentity.SessionID, + Content: "already settled", QueuedBy: QueuedByUser, + } + if err := repo.InsertForSession(ctx, destinationIdentity, destinationPrime, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + primeClaim, err := repo.ClaimSendNowForSession(ctx, destinationIdentity, []QueuedMessage{*destinationPrime}) + if err != nil { + t.Fatal(err) + } + if err := repo.AcknowledgeSendNowClaim(ctx, primeClaim); err != nil { + t.Fatal(err) + } + source := &QueuedMessage{ + ID: "source", TaskID: sourceIdentity.TaskID, SessionID: sourceIdentity.SessionID, + Content: "durable source", QueuedBy: QueuedByWorkflow, + Metadata: map[string]interface{}{MetadataLifecycleDurable: true}, + } + if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + claim, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source}) + if err != nil { + t.Fatal(err) + } + sourceGeneration := claim.OperationGeneration + if accepted { + if err := repo.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil { + t.Fatal(err) + } + } + + if err := repo.TransferSessionIdentities(ctx, sourceIdentity, destinationIdentity); err != nil { + t.Fatal(err) + } + + pending, err := repo.ListPendingSendNowClaims(ctx) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 { + t.Fatalf("pending claims = %#v, want one transferred claim", pending) + } + transferred := pending[0] + if transferred.Claim.Identity != destinationIdentity { + t.Fatalf("transferred claim identity = %#v, want %#v", transferred.Claim.Identity, destinationIdentity) + } + if transferred.Claim.OperationGeneration != transferred.Claim.SessionGeneration { + t.Fatalf("transferred claim generations = operation %d, session %d", transferred.Claim.OperationGeneration, transferred.Claim.SessionGeneration) + } + if transferred.Claim.OperationGeneration <= sourceGeneration { + t.Fatalf("transferred operation generation = %d, want greater than source %d", transferred.Claim.OperationGeneration, sourceGeneration) + } + if accepted { + err = repo.AcknowledgeSendNowClaim(ctx, &transferred.Claim) + } else { + err = repo.RestoreSendNowClaim(ctx, &transferred.Claim) + } + if err != nil { + t.Fatal(err) + } + }) + } +} + +func TestTransferredSendNowClaimFencesOriginalSourceActions(t *testing.T) { + transfers := []struct { + name string + run func(context.Context, *sqliteRepository, QueueSessionIdentity, QueueSessionIdentity) error + }{ + { + name: "direct", + run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error { + return repo.TransferSessionIdentities(ctx, source, destination) + }, + }, + { + name: "durable", + run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error { + service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession) + return service.TransferSessionWithDurableAttachmentPreparation( + ctx, source.TaskID, source.SessionID, destination.SessionID, nil, nil, + ) + }, + }, + } + actions := []struct { + name string + run func(context.Context, *sqliteRepository, *SendNowClaim) error + }{ + {name: "restore", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error { + return repo.RestoreSendNowClaim(ctx, claim) + }}, + {name: "acknowledge", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error { + return repo.AcknowledgeSendNowClaim(ctx, claim) + }}, + {name: "accept", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error { + return repo.MarkPendingSendNowClaimAccepted(ctx, claim) + }}, + {name: "startup discard", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error { + return repo.DeletePendingSendNowClaim(ctx, claim) + }}, + } + + for _, transfer := range transfers { + for _, accepted := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/accepted=%t", transfer.name, accepted), func(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t).(*sqliteRepository) + sourceIdentity := QueueSessionIdentity{ + TaskID: "task-1", + SessionID: "session-old", + SessionIncarnationID: "incarnation-old", + } + destinationIdentity := QueueSessionIdentity{ + TaskID: sourceIdentity.TaskID, + SessionID: "session-new", + SessionIncarnationID: "incarnation-new", + } + seedQueueSessionIdentity(t, repo, sourceIdentity) + seedQueueSessionIdentity(t, repo, destinationIdentity) + for _, source := range []*QueuedMessage{ + {ID: "first", TaskID: sourceIdentity.TaskID, SessionID: sourceIdentity.SessionID, Content: "first", QueuedBy: QueuedByUser}, + {ID: "second", TaskID: sourceIdentity.TaskID, SessionID: sourceIdentity.SessionID, Content: "second", QueuedBy: QueuedByUser}, + } { + if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil { + t.Fatal(err) + } + } + sources, err := repo.ListBySession(ctx, sourceIdentity.SessionID) + if err != nil { + t.Fatal(err) + } + claim, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, sources) + if err != nil { + t.Fatal(err) + } + if accepted { + if err := repo.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil { + t.Fatal(err) + } + } + if err := transfer.run(ctx, repo, sourceIdentity, destinationIdentity); err != nil { + t.Fatal(err) + } + + for _, action := range actions { + t.Run(action.name, func(t *testing.T) { + err := action.run(ctx, repo, claim) + if !errors.Is(err, ErrSendNowClaimChanged) && !errors.Is(err, ErrSessionIdentityMismatch) { + t.Fatalf("old source %s = %v, want claim fencing error", action.name, err) + } + pending, err := repo.ListPendingSendNowClaims(ctx) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 { + t.Fatalf("pending claims after stale %s = %#v, want successor claim", action.name, pending) + } + transferred := pending[0] + if transferred.Claim.ClaimID != claim.ClaimID || transferred.Claim.Identity != destinationIdentity || transferred.Accepted != accepted { + t.Fatalf("transferred claim after stale %s = %#v", action.name, transferred) + } + if len(transferred.Claim.Sources) != 2 || + transferred.Claim.Sources[0].Content != "first" || + transferred.Claim.Sources[1].Content != "second" || + transferred.Claim.Sources[0].SessionID != destinationIdentity.SessionID || + transferred.Claim.Sources[1].SessionID != destinationIdentity.SessionID { + t.Fatalf("transferred FIFO sources after stale %s = %#v", action.name, transferred.Claim.Sources) + } + }) + } + }) + } + } +} + func TestSQLitePurgeRemovesDurableSendNowClaims(t *testing.T) { for _, purge := range []struct { name string diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index 73bafe7099..48e4d91381 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go @@ -5529,8 +5529,9 @@ func lifecycleReservationFromMetadataJSON(metadataJSON string) (string, bool, er func (r *sqliteRepository) transferSessionOwned( ctx context.Context, oldSessionID, newSessionID, operationID string, + source, destination *QueueSessionIdentity, ) error { - return r.transferSessionOwnedTx(ctx, oldSessionID, newSessionID, operationID) + return r.transferSessionOwnedTx(ctx, oldSessionID, newSessionID, operationID, source, destination) } func (r *sqliteRepository) beginAuthorizedSessionTransferTx( @@ -5643,30 +5644,32 @@ func (r *sqliteRepository) transferSessionQueueRowsTx( ctx context.Context, tx *sqlx.Tx, oldSessionID, newSessionID string, -) error { +) (int64, error) { var destinationMax sql.NullInt64 if err := tx.GetContext(ctx, &destinationMax, r.db.Rebind(` SELECT MAX(position) FROM queued_messages WHERE session_id = ? `), newSessionID); err != nil { - return fmt.Errorf("transfer max: %w", err) + return 0, fmt.Errorf("transfer max: %w", err) } if _, err := tx.ExecContext(ctx, r.db.Rebind(` UPDATE queued_messages SET session_id = ?, position = position + ? WHERE session_id = ? `), newSessionID, destinationMax.Int64, oldSessionID); err != nil { - return fmt.Errorf("transfer queued: %w", err) + return 0, fmt.Errorf("transfer queued: %w", err) } var transferredMax sql.NullInt64 if err := tx.GetContext(ctx, &transferredMax, r.db.Rebind(` SELECT MAX(position) FROM queued_messages WHERE session_id = ? `), newSessionID); err != nil { - return fmt.Errorf("transfer destination max after move: %w", err) + return 0, fmt.Errorf("transfer destination max after move: %w", err) } if transferredMax.Valid { - return r.bumpQueuePositionTx(ctx, tx, newSessionID, transferredMax.Int64) + if err := r.bumpQueuePositionTx(ctx, tx, newSessionID, transferredMax.Int64); err != nil { + return 0, err + } } - return nil + return destinationMax.Int64, nil } func (r *sqliteRepository) transferSessionStateTx( @@ -5745,6 +5748,13 @@ func (r *sqliteRepository) transferSession( if err := r.lockAndValidateTransferSessionsTx(ctx, tx, source, destination, first, second); err != nil { return err } + sourceGeneration := int64(0) + if source != nil { + sourceGeneration, err = r.getSendNowGenerationTx(ctx, tx, oldSessionID) + if err != nil { + return err + } + } if oldSessionID == newSessionID { return tx.Commit() } @@ -5770,7 +5780,8 @@ func (r *sqliteRepository) transferSession( if err := r.deleteEditLeasesForTransferTx(ctx, tx, oldSessionID, newSessionID); err != nil { return err } - if err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID); err != nil { + queuePositionOffset, err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID) + if err != nil { return err } if _, err := tx.ExecContext(ctx, r.db.Rebind(`DELETE FROM pending_moves WHERE session_id = ?`), newSessionID); err != nil { @@ -5791,7 +5802,9 @@ func (r *sqliteRepository) transferSession( if err := r.bumpSendNowGenerationTx(ctx, tx, newSessionID); err != nil { return err } - if err := r.transferPendingSendNowClaimTx(ctx, tx, oldSessionID, newSessionID); err != nil { + if err := r.transferPendingSendNowClaimTx( + ctx, tx, oldSessionID, newSessionID, source, destination, sourceGeneration, queuePositionOffset, + ); err != nil { return err } return tx.Commit() @@ -5800,6 +5813,7 @@ func (r *sqliteRepository) transferSession( func (r *sqliteRepository) transferSessionOwnedTx( ctx context.Context, oldSessionID, newSessionID, operationID string, + source, destination *QueueSessionIdentity, ) error { if err := r.ensureQueueDispatchRecoverySchema(ctx); err != nil { return err @@ -5834,6 +5848,23 @@ func (r *sqliteRepository) transferSessionOwnedTx( return err } defer func() { _ = tx.Rollback() }() + if source != nil { + if err := r.validateSessionIdentityTx(ctx, tx, *source); err != nil { + return err + } + } + if destination != nil { + if err := r.validateSessionIdentityTx(ctx, tx, *destination); err != nil { + return err + } + } + sourceGeneration := int64(0) + if source != nil { + sourceGeneration, err = r.getSendNowGenerationTx(ctx, tx, oldSessionID) + if err != nil { + return err + } + } if oldSessionID == newSessionID { return commitAuthorizedSessionTransferTx( ctx, tx, r.db, oldSessionID, newSessionID, operationID, @@ -5845,13 +5876,16 @@ func (r *sqliteRepository) transferSessionOwnedTx( if err := r.deleteEditLeasesForTransferTx(ctx, tx, oldSessionID, newSessionID); err != nil { return err } - if err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID); err != nil { + queuePositionOffset, err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID) + if err != nil { return err } if err := r.transferSessionStateTx(ctx, tx, oldSessionID, newSessionID); err != nil { return err } - if err := r.transferPendingSendNowClaimTx(ctx, tx, oldSessionID, newSessionID); err != nil { + if err := r.transferPendingSendNowClaimTx( + ctx, tx, oldSessionID, newSessionID, source, destination, sourceGeneration, queuePositionOffset, + ); err != nil { return err } return commitAuthorizedSessionTransferTx( diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index c03ae80bdb..2efb1579c1 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -3373,12 +3373,13 @@ func (s *Service) SnapshotSession(ctx context.Context, sessionID string) ([]Queu } type ownedSessionTransferRepository interface { - transferSessionOwned(context.Context, string, string, string) error + transferSessionOwned(context.Context, string, string, string, *QueueSessionIdentity, *QueueSessionIdentity) error } func (s *Service) transferRepositorySession( ctx context.Context, oldSessionID, newSessionID, operationID string, + source, destination *QueueSessionIdentity, ) error { if operationID == "" { return s.repo.TransferSession(ctx, oldSessionID, newSessionID) @@ -3387,7 +3388,7 @@ func (s *Service) transferRepositorySession( if !ok { return errors.New("owned session transfer unavailable") } - return repo.transferSessionOwned(ctx, oldSessionID, newSessionID, operationID) + return repo.transferSessionOwned(ctx, oldSessionID, newSessionID, operationID, source, destination) } // SnapshotSessionForIdentity captures queue and deferred-move state for one exact incarnation. @@ -3473,25 +3474,45 @@ func (s *Service) transferRepositorySessionForTask( ctx context.Context, taskID, oldSessionID, newSessionID, operationID string, ) error { + transferred, err := s.transferRepositorySessionWithLiveIdentities( + ctx, taskID, oldSessionID, newSessionID, operationID, + ) + if err != nil { + return err + } + if transferred { + return nil + } + return s.transferRepositorySession(ctx, oldSessionID, newSessionID, operationID, nil, nil) +} + +func (s *Service) transferRepositorySessionWithLiveIdentities( + ctx context.Context, + taskID, oldSessionID, newSessionID, operationID string, +) (bool, error) { // Transfers that do not have a durable compensation record can use the // immutable session identities. This fences a workflow handoff to one task // and lets repositories reject a stale or cross-task destination. Durable // transfers keep the owned-operation path because it also carries the // compensation lease and operation token. - if taskID != "" && operationID == "" { - source, sourceErr := s.repo.ResolveSessionIdentity(ctx, taskID, oldSessionID) - if sourceErr != nil && !errors.Is(sourceErr, ErrSessionIdentityMismatch) { - return sourceErr - } - destination, destinationErr := s.repo.ResolveSessionIdentity(ctx, taskID, newSessionID) - if destinationErr != nil && !errors.Is(destinationErr, ErrSessionIdentityMismatch) { - return destinationErr - } - if sourceErr == nil && destinationErr == nil { - return s.repo.TransferSessionIdentities(ctx, source, destination) - } + if taskID == "" { + return false, nil } - return s.transferRepositorySession(ctx, oldSessionID, newSessionID, operationID) + source, sourceErr := s.repo.ResolveSessionIdentity(ctx, taskID, oldSessionID) + if sourceErr != nil && !errors.Is(sourceErr, ErrSessionIdentityMismatch) { + return false, sourceErr + } + destination, destinationErr := s.repo.ResolveSessionIdentity(ctx, taskID, newSessionID) + if destinationErr != nil && !errors.Is(destinationErr, ErrSessionIdentityMismatch) { + return false, destinationErr + } + if sourceErr != nil || destinationErr != nil { + return false, nil + } + if operationID == "" { + return true, s.repo.TransferSessionIdentities(ctx, source, destination) + } + return true, s.transferRepositorySession(ctx, oldSessionID, newSessionID, operationID, &source, &destination) } type attachmentCleanupRepository interface { diff --git a/apps/backend/internal/orchestrator/messagequeue/service_session_transfer.go b/apps/backend/internal/orchestrator/messagequeue/service_session_transfer.go index e82f2ea697..63f934f7e1 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service_session_transfer.go +++ b/apps/backend/internal/orchestrator/messagequeue/service_session_transfer.go @@ -170,6 +170,14 @@ func (s *Service) prepareDurableSessionTransfer( } entryIDs = append(entryIDs, dispatchEntryIDs...) attachmentIDs = append(attachmentIDs, dispatchAttachments...) + claimEntryIDs, claimAttachments, err := s.snapshotPendingSendNowClaimTransfer( + ctx, taskID, oldSessionID, seenEntryIDs, seenAttachmentIDs, + ) + if err != nil { + return err + } + entryIDs = append(entryIDs, claimEntryIDs...) + attachmentIDs = append(attachmentIDs, claimAttachments...) cleanupEntryIDs, cleanupAttachments, cleanupLocators, err := s.snapshotAttachmentCleanupTransfer( ctx, taskID, oldSessionID, seenEntryIDs, seenAttachmentIDs, ) @@ -208,6 +216,43 @@ func (s *Service) prepareDurableSessionTransfer( return nil } +func (s *Service) snapshotPendingSendNowClaimTransfer( + ctx context.Context, + taskID, sessionID string, + seenEntryIDs, seenAttachmentIDs map[string]struct{}, +) ([]string, []string, error) { + if !s.PendingSendNowClaimPersistenceAvailable() { + return nil, nil, nil + } + claims, err := s.ListPendingSendNowClaims(ctx) + if err != nil { + return nil, nil, fmt.Errorf("snapshot pending Send Now claims for transfer: %w", err) + } + entryIDs := make([]string, 0) + attachmentIDs := make([]string, 0) + for _, pending := range claims { + claim := pending.Claim + if claim.Dispatch.SessionID != sessionID || + (taskID != "" && claim.Dispatch.TaskID != "" && claim.Dispatch.TaskID != taskID) { + continue + } + for _, source := range claim.Sources { + if source.SessionID != sessionID || (taskID != "" && source.TaskID != taskID) { + continue + } + if source.ID != "" { + if _, seen := seenEntryIDs[source.ID]; !seen { + entryIDs = append(entryIDs, source.ID) + seenEntryIDs[source.ID] = struct{}{} + } + } + attachmentIDs = appendTransferAttachmentIDs(attachmentIDs, seenAttachmentIDs, source.Attachments) + } + attachmentIDs = appendTransferAttachmentIDs(attachmentIDs, seenAttachmentIDs, claim.Dispatch.Attachments) + } + return entryIDs, attachmentIDs, nil +} + func (s *Service) snapshotPendingDispatchTransfer( ctx context.Context, sessionID string, diff --git a/apps/backend/internal/orchestrator/messagequeue/service_transfer_compensation_test.go b/apps/backend/internal/orchestrator/messagequeue/service_transfer_compensation_test.go index e6ba207c72..190b7b414c 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service_transfer_compensation_test.go +++ b/apps/backend/internal/orchestrator/messagequeue/service_transfer_compensation_test.go @@ -3,9 +3,11 @@ package messagequeue import ( "context" "errors" + "fmt" + "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "testing" ) func TestTransferSessionWithPreparationRollsBackPreparationError(t *testing.T) { @@ -126,6 +128,100 @@ func TestDurableSessionTransferIncludesInFlightOrdinaryAttachment(t *testing.T) assert.Equal(t, queued.ID, pending[0].Message.ID) } +func TestDurableSessionTransferRebindsPendingSendNowClaimIdentity(t *testing.T) { + for _, accepted := range []bool{false, true} { + t.Run(fmt.Sprintf("accepted=%t", accepted), func(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t) + sourceIdentity := QueueSessionIdentity{ + TaskID: "task", + SessionID: "session-old", + SessionIncarnationID: "incarnation-old", + } + destinationIdentity := QueueSessionIdentity{ + TaskID: sourceIdentity.TaskID, + SessionID: "session-new", + SessionIncarnationID: "incarnation-new", + } + seedQueueSessionIdentity(t, repo, sourceIdentity) + seedQueueSessionIdentity(t, repo, destinationIdentity) + service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession) + source, err := service.QueueMessageWithMetadataForSession( + ctx, sourceIdentity, "handoff", "", "user", false, nil, nil, + ) + require.NoError(t, err) + claim, err := service.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source}) + require.NoError(t, err) + if accepted { + require.NoError(t, service.MarkPendingSendNowClaimAccepted(ctx, claim)) + } + + require.NoError(t, service.TransferSessionWithDurableAttachmentPreparation( + ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil, + )) + pending, err := service.ListPendingSendNowClaims(ctx) + require.NoError(t, err) + require.Len(t, pending, 1) + assert.Equal(t, destinationIdentity, pending[0].Claim.Identity) + assert.Equal(t, pending[0].Claim.SessionGeneration, pending[0].Claim.OperationGeneration) + assert.Equal(t, accepted, pending[0].Accepted) + }) + } +} + +func TestDurableSessionTransferRestoresPendingSendNowClaimAfterDestinationQueue(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t) + service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession) + service.SetAutoMergeEnabled(false) + sourceIdentity := QueueSessionIdentity{ + TaskID: "task", + SessionID: "session-old", + SessionIncarnationID: "incarnation-old", + } + destinationIdentity := QueueSessionIdentity{ + TaskID: sourceIdentity.TaskID, + SessionID: "session-new", + SessionIncarnationID: "incarnation-new", + } + seedQueueSessionIdentity(t, repo, sourceIdentity) + seedQueueSessionIdentity(t, repo, destinationIdentity) + _, err := service.QueueMessageWithMetadataForSession( + ctx, destinationIdentity, "destination", "", QueuedByUser, false, nil, nil, + ) + require.NoError(t, err) + first, err := service.QueueMessageWithMetadataForSession( + ctx, sourceIdentity, "first", "", QueuedByUser, false, nil, nil, + ) + require.NoError(t, err) + second, err := service.QueueMessageWithMetadataForSession( + ctx, sourceIdentity, "second", "", QueuedByUser, false, nil, nil, + ) + require.NoError(t, err) + sources, err := repo.ListBySession(ctx, sourceIdentity.SessionID) + require.NoError(t, err) + require.Equal(t, []string{first.ID, second.ID}, []string{sources[0].ID, sources[1].ID}) + claim, err := service.ClaimSendNowForSession(ctx, sourceIdentity, sources) + require.NoError(t, err) + + require.NoError(t, service.TransferSessionWithDurableAttachmentPreparation( + ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil, + )) + pending, err := service.ListPendingSendNowClaims(ctx) + require.NoError(t, err) + require.Len(t, pending, 1) + require.Equal(t, claim.ClaimID, pending[0].Claim.ClaimID) + require.NoError(t, service.RestoreSendNowClaim(ctx, &pending[0].Claim)) + + entries, err := repo.ListBySession(ctx, destinationIdentity.SessionID) + require.NoError(t, err) + require.Equal(t, []string{"destination", "first", "second"}, []string{ + entries[0].Content, entries[1].Content, entries[2].Content, + }) + require.Less(t, entries[0].Position, entries[1].Position) + require.Less(t, entries[1].Position, entries[2].Position) +} + func TestDurableSessionTransferIncludesCleanupOnlyAttachmentClaim(t *testing.T) { ctx := context.Background() repo := newTestSQLiteRepo(t) diff --git a/apps/backend/internal/orchestrator/messagequeue/service_transfer_send_now_attachments_test.go b/apps/backend/internal/orchestrator/messagequeue/service_transfer_send_now_attachments_test.go new file mode 100644 index 0000000000..a317cec62a --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/service_transfer_send_now_attachments_test.go @@ -0,0 +1,56 @@ +package messagequeue + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestDurableSessionTransferIncludesPendingSendNowClaimAttachments(t *testing.T) { + ctx := context.Background() + repo := newTestSQLiteRepo(t) + service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession) + source, err := service.QueueMessage( + ctx, + "session-old", + "task", + "handoff", + "", + QueuedByUser, + false, + []MessageAttachment{{AttachmentID: "claim-attachment"}}, + ) + require.NoError(t, err) + _, err = service.ClaimSendNow(ctx, source.SessionID, []QueuedMessage{*source}) + require.NoError(t, err) + + var preparedAttachments []string + err = service.TransferSessionWithDurableAttachmentPreparation( + ctx, + "task", + "session-old", + "session-new", + func(_ context.Context, attachmentIDs []string) error { + preparedAttachments = append(preparedAttachments, attachmentIDs...) + compensations, listErr := service.ListSessionTransferCompensations(ctx) + require.NoError(t, listErr) + require.Len(t, compensations, 1) + require.Equal(t, []string{"claim-attachment"}, compensations[0].AttachmentIDs) + return nil + }, + nil, + ) + require.NoError(t, err) + require.Equal(t, []string{"claim-attachment"}, preparedAttachments) + + pending, err := service.ListPendingSendNowClaims(ctx) + require.NoError(t, err) + require.Len(t, pending, 1) + require.Equal(t, "session-new", pending[0].Claim.Dispatch.SessionID) + require.NoError(t, service.RestoreSendNowClaim(ctx, &pending[0].Claim)) + entries, err := repo.ListBySession(ctx, "session-new") + require.NoError(t, err) + require.Len(t, entries, 1) + require.Equal(t, "claim-attachment", entries[0].Attachments[0].AttachmentID) +} diff --git a/apps/backend/internal/orchestrator/queue_send_now_recovery_test.go b/apps/backend/internal/orchestrator/queue_send_now_recovery_test.go index 114ed86704..309650e655 100644 --- a/apps/backend/internal/orchestrator/queue_send_now_recovery_test.go +++ b/apps/backend/internal/orchestrator/queue_send_now_recovery_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "fmt" "path/filepath" "sync/atomic" @@ -111,6 +112,103 @@ func TestSendNowAcceptedClaimIsAcknowledgedAfterProcessRestart(t *testing.T) { } } +func TestTransferredIdentityAwareSendNowClaimReconcilesAfterProcessRestart(t *testing.T) { + for _, accepted := range []bool{false, true} { + t.Run(fmt.Sprintf("accepted=%t", accepted), func(t *testing.T) { + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "queue.db") + queue, db, sourceIdentity, destinationIdentity := newIdentityWorkflowTransferQueue(t, dbPath, true) + source, err := queue.QueueMessageWithMetadataForSession( + ctx, sourceIdentity, "transferred prompt", "", "user", false, nil, nil, + ) + if err != nil { + t.Fatal(err) + } + claim, err := queue.ClaimSendNowForSession(ctx, sourceIdentity, []messagequeue.QueuedMessage{*source}) + if err != nil { + t.Fatal(err) + } + if accepted { + if err := queue.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil { + t.Fatal(err) + } + } + if err := queue.TransferSessionWithDurableAttachmentPreparation( + ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil, + ); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`DELETE FROM task_sessions WHERE id = ?`, sourceIdentity.SessionID); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + + restartedQueue, restartedDB, _, _ := newIdentityWorkflowTransferQueue(t, dbPath, false) + t.Cleanup(func() { _ = restartedDB.Close() }) + restarted := &Service{logger: testLogger(), messageQueue: restartedQueue} + if err := restarted.reconcilePendingSendNowClaimsOnStartup(ctx); err != nil { + t.Fatal(err) + } + status := restartedQueue.GetStatus(ctx, destinationIdentity.SessionID) + if accepted && status.Count != 0 { + t.Fatalf("accepted transferred claim restored after restart: %#v", status.Entries) + } + if !accepted && (len(status.Entries) != 1 || status.Entries[0].ID != source.ID) { + t.Fatalf("unaccepted transferred claim after restart = %#v, want source %s", status.Entries, source.ID) + } + }) + } +} + +func newIdentityWorkflowTransferQueue( + t *testing.T, + dbPath string, + seed bool, +) (*messagequeue.Service, *sqlx.DB, messagequeue.QueueSessionIdentity, messagequeue.QueueSessionIdentity) { + t.Helper() + raw, err := sql.Open("sqlite3", dbPath) + if err != nil { + t.Fatal(err) + } + raw.SetMaxOpenConns(1) + db := sqlx.NewDb(raw, "sqlite3") + statements := []string{ + `CREATE TABLE IF NOT EXISTS tasks (id TEXT PRIMARY KEY, archived_at TIMESTAMP, updated_at TIMESTAMP)`, + `CREATE TABLE IF NOT EXISTS task_sessions (id TEXT PRIMARY KEY, task_id TEXT NOT NULL, queue_incarnation_id TEXT NOT NULL)`, + } + for _, statement := range statements { + if _, err := db.Exec(statement); err != nil { + _ = db.Close() + t.Fatal(err) + } + } + source := messagequeue.QueueSessionIdentity{TaskID: "task-1", SessionID: "session-old", SessionIncarnationID: "incarnation-old"} + destination := messagequeue.QueueSessionIdentity{TaskID: source.TaskID, SessionID: "session-new", SessionIncarnationID: "incarnation-new"} + if seed { + if _, err := db.Exec(`INSERT OR IGNORE INTO tasks (id, updated_at) VALUES (?, CURRENT_TIMESTAMP)`, source.TaskID); err != nil { + _ = db.Close() + t.Fatal(err) + } + for _, identity := range []messagequeue.QueueSessionIdentity{source, destination} { + if _, err := db.Exec(` + INSERT INTO task_sessions (id, task_id, queue_incarnation_id) VALUES (?, ?, ?) + ON CONFLICT(id) DO UPDATE SET task_id = excluded.task_id, queue_incarnation_id = excluded.queue_incarnation_id + `, identity.SessionID, identity.TaskID, identity.SessionIncarnationID); err != nil { + _ = db.Close() + t.Fatal(err) + } + } + } + repo, err := messagequeue.NewSQLiteRepository(db, db) + if err != nil { + _ = db.Close() + t.Fatal(err) + } + return messagequeue.NewService(repo, messagequeue.DefaultMaxPerSession, testLogger()), db, source, destination +} + type transientAcceptedMarkerRepository struct { messagequeue.Repository pending interface { diff --git a/docs/plans/send-now-claim-identity/plan.md b/docs/plans/send-now-claim-identity/plan.md new file mode 100644 index 0000000000..87ddb19828 --- /dev/null +++ b/docs/plans/send-now-claim-identity/plan.md @@ -0,0 +1,34 @@ +--- +created: 2026-09-21 +status: complete +requirements: + - REQ-UI-MESSAGE-QUEUE-SEND-NOW-001 +system_design: + - ../../specs/ui/system-design/message-queue-send-now.md +legacy_specs: [] +--- + +# Implementation Plan: Preserve Send Now claim identity across session handoff + +Session handoff must carry pending Send Now claims to the successor session +without carrying the retired session's incarnation or operation generation. +This corrective package records the narrow persistence repair and its restart +reconciliation proof. The existing Send Now requirement and system design +remain authoritative for queue ownership, FIFO ordering, and fencing. + +## Scope + +- Rebind accepted and unaccepted pending claims to the live destination + session identity during both direct and durable owned transfers. +- Preserve claim IDs, accepted state, source ordering, dispatch metadata, + transaction rollback, and stale source fencing. +- Prove recovery after the source session is removed and the process restarts. + +## Out of scope + +Queue feature behavior, UI changes, provider routing, schema changes, incident +database repair, deployment, and backup cleanup. + +## Delivery + +See [Task 01: Rebind transferred Send Now claims](task-01-rebind-claims.md). diff --git a/docs/plans/send-now-claim-identity/task-01-rebind-claims.md b/docs/plans/send-now-claim-identity/task-01-rebind-claims.md new file mode 100644 index 0000000000..8a91aafca7 --- /dev/null +++ b/docs/plans/send-now-claim-identity/task-01-rebind-claims.md @@ -0,0 +1,56 @@ +--- +id: "01-rebind-claims" +title: "Rebind transferred Send Now claims" +status: complete +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-UI-MESSAGE-QUEUE-SEND-NOW-001 +acceptance_criteria: + - AC-UI-MESSAGE-QUEUE-SEND-NOW-001.8 + - AC-UI-MESSAGE-QUEUE-SEND-NOW-001.10 +system_design: + - ../../specs/ui/system-design/message-queue-send-now.md +--- + +# Task 01: Rebind transferred Send Now claims + +## Summary + +When a session handoff moves a pending Send Now claim, persist the successor's +full queue session identity and operation generation. This keeps restart +reconciliation usable while retaining exact claim fencing. + +## Acceptance + +- Accepted and unaccepted claims transferred through direct and durable owned + paths use the destination session incarnation and generation. +- Claim IDs, accepted state, source rows, dispatch metadata, FIFO order, and + rollback behavior remain unchanged. +- Requests and settlements using the retired source identity remain fenced. +- Legacy transfer fallback fails closed when it encounters an identity-aware + claim instead of persisting the retired identity. +- Reconciliation after source-session removal and process restart can settle + both claim states under the successor identity. + +## Verification + +```bash +(cd apps/backend && go test -v -run '^(TestSQLiteTransferSessionIdentitiesRebindsPendingSendNowClaim|TestTransferredIdentityAwareSendNowClaimReconcilesAfterProcessRestart)$' ./internal/orchestrator/messagequeue ./internal/orchestrator) +(cd apps/backend && go test -race ./internal/orchestrator/messagequeue) +(cd apps/backend && go test ./internal/orchestrator/messagequeue ./internal/orchestrator) +``` + +## Results + +The direct and durable transfer paths now pass the live destination identity +into the atomic transfer transaction, which validates it before rebinding the +claim and derives the claim generation from the destination session generation. +Durable transfer preparation also includes attachments held only by a pending +Send Now claim after the source queue row is removed. The legacy transfer +fallback rejects an identity-aware claim rather than retaining the retired +identity, while identity-free legacy claims keep their existing behavior. +Accepted and unaccepted transfer, restart reconciliation, source fencing, +fallback rejection, claim attachment preparation, and race coverage pass. No +UI or public documentation change is required.