diff --git a/services/game-service/internal/storage/mysql/outbox_claim_priority_test.go b/services/game-service/internal/storage/mysql/outbox_claim_priority_test.go new file mode 100644 index 00000000..eea0668f --- /dev/null +++ b/services/game-service/internal/storage/mysql/outbox_claim_priority_test.go @@ -0,0 +1,81 @@ +package mysql + +import ( + "context" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "hyapp/pkg/appcode" +) + +func TestClaimPendingGameOutboxPrioritizesRecoveryOverPendingAtBatchOne(t *testing.T) { + tests := []struct { + name string + eventID string + recoveryStatus string + expectEmptyRetry bool + }{ + {name: "due retry", eventID: "game-retry", recoveryStatus: "retryable"}, + {name: "expired run", eventID: "game-expired", recoveryStatus: "running", expectEmptyRetry: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("create sqlmock: %v", err) + } + defer db.Close() + + mock.ExpectBegin() + if tt.expectEmptyRetry { + mock.ExpectQuery(gameOutboxRetryClaimPattern()). + WithArgs("lalu", sqlmock.AnyArg(), 1). + WillReturnRows(gameOutboxClaimRows()) + } + claimPattern := gameOutboxRetryClaimPattern() + if tt.recoveryStatus == "running" { + claimPattern = gameOutboxExpiredClaimPattern() + } + mock.ExpectQuery(claimPattern). + WithArgs("lalu", sqlmock.AnyArg(), 1). + WillReturnRows(gameOutboxClaimRows().AddRow( + "lalu", tt.eventID, "GameOrderSettled", "order-1", int64(1001), "self", "dice", "debit", + int64(10), `{}`, int64(200), + )) + mock.ExpectExec(`(?s)UPDATE game_outbox\s+SET status = 'running', worker_id = \?, lock_until_ms = \?, updated_at_ms = \?\s+WHERE app_code = \? AND event_id = \?`). + WithArgs("worker-1", sqlmock.AnyArg(), sqlmock.AnyArg(), "lalu", tt.eventID). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + + records, err := (&Repository{db: db}).ClaimPendingGameOutbox( + appcode.WithContext(context.Background(), "lalu"), "worker-1", 1, + ) + if err != nil { + t.Fatalf("claim game recovery row: %v", err) + } + if len(records) != 1 || records[0].EventID != tt.eventID { + t.Fatalf("batch=1 must claim %s before an older pending row: %+v", tt.recoveryStatus, records) + } + // No pending query is registered: issuing one after the recovery row filled + // the batch makes sqlmock fail, protecting the starvation boundary directly. + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("game claim priority mismatch: %v", err) + } + }) + } +} + +func gameOutboxClaimRows() *sqlmock.Rows { + return sqlmock.NewRows([]string{ + "app_code", "event_id", "event_type", "order_id", "user_id", "platform_code", "game_id", "op_type", + "coin_amount", "payload_json", "created_at_ms", + }) +} + +func gameOutboxRetryClaimPattern() string { + return `(?s)FROM game_outbox FORCE INDEX \(idx_game_outbox_status_created\).*WHERE app_code = \? AND status = 'retryable' AND next_retry_at_ms <= \? AND lock_until_ms = 0.*FOR UPDATE SKIP LOCKED` +} + +func gameOutboxExpiredClaimPattern() string { + return `(?s)FROM game_outbox FORCE INDEX \(idx_game_outbox_lock\).*WHERE app_code = \? AND status = 'running' AND lock_until_ms <= \?.*FOR UPDATE SKIP LOCKED` +} diff --git a/services/game-service/internal/storage/mysql/repository.go b/services/game-service/internal/storage/mysql/repository.go index 62f6ba0a..8386fe0b 100644 --- a/services/game-service/internal/storage/mysql/repository.go +++ b/services/game-service/internal/storage/mysql/repository.go @@ -1196,21 +1196,14 @@ func (r *Repository) ClaimPendingGameOutbox(ctx context.Context, workerID string lockUntilMS := nowMS + (30 * time.Second).Milliseconds() appCode := appcode.FromContext(ctx) records := make([]GameOutboxRecord, 0, batchSize) + // Recovery work must claim before fresh pending facts. A continuously growing + // pending backlog can otherwise fill every batch and permanently strand both + // due retries and running rows whose worker lease expired. Keep the states in + // separate SKIP LOCKED queries so each branch retains its status-leading index. claimBranches := []struct { query string args []any }{ - { - query: ` - SELECT app_code, event_id, event_type, order_id, user_id, platform_code, game_id, op_type, - coin_amount, CAST(payload_json AS CHAR), created_at_ms - FROM game_outbox FORCE INDEX (idx_game_outbox_status_created) - WHERE app_code = ? AND status = 'pending' AND next_retry_at_ms <= ? AND lock_until_ms = 0 - ORDER BY created_at_ms ASC, event_id ASC - LIMIT ? FOR UPDATE SKIP LOCKED - `, - args: []any{appCode, nowMS}, - }, { query: ` SELECT app_code, event_id, event_type, order_id, user_id, platform_code, game_id, op_type, @@ -1230,6 +1223,17 @@ func (r *Repository) ClaimPendingGameOutbox(ctx context.Context, workerID string WHERE app_code = ? AND status = 'running' AND lock_until_ms <= ? ORDER BY created_at_ms ASC, event_id ASC LIMIT ? FOR UPDATE SKIP LOCKED + `, + args: []any{appCode, nowMS}, + }, + { + query: ` + SELECT app_code, event_id, event_type, order_id, user_id, platform_code, game_id, op_type, + coin_amount, CAST(payload_json AS CHAR), created_at_ms + FROM game_outbox FORCE INDEX (idx_game_outbox_status_created) + WHERE app_code = ? AND status = 'pending' AND next_retry_at_ms <= ? AND lock_until_ms = 0 + ORDER BY created_at_ms ASC, event_id ASC + LIMIT ? FOR UPDATE SKIP LOCKED `, args: []any{appCode, nowMS}, }, diff --git a/services/user-service/internal/storage/mysql/user_outbox.go b/services/user-service/internal/storage/mysql/user_outbox.go index b77e6a9a..005f57ab 100644 --- a/services/user-service/internal/storage/mysql/user_outbox.go +++ b/services/user-service/internal/storage/mysql/user_outbox.go @@ -69,20 +69,14 @@ func (r *Repository) ClaimPendingUserOutbox(ctx context.Context, workerID string lockUntilMS := nowMS + (30 * time.Second).Milliseconds() appCode := appcode.FromContext(ctx) records := make([]UserOutboxRecord, 0, batchSize) + // Recovery work must claim before fresh pending facts. A continuously growing + // pending backlog can otherwise fill every batch and permanently strand both + // due retries and running rows whose worker lease expired. Keep the states in + // separate SKIP LOCKED queries so each branch retains its status-leading index. claimBranches := []struct { query string args []any }{ - { - query: ` - SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms - FROM user_outbox FORCE INDEX (idx_user_outbox_status_created) - WHERE app_code = ? AND status = 'pending' AND next_retry_at_ms <= ? AND lock_until_ms = 0 - ORDER BY created_at_ms ASC, event_id ASC - LIMIT ? FOR UPDATE SKIP LOCKED - `, - args: []any{appCode, nowMS}, - }, { query: ` SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms @@ -100,6 +94,16 @@ func (r *Repository) ClaimPendingUserOutbox(ctx context.Context, workerID string WHERE app_code = ? AND status = 'running' AND lock_until_ms <= ? ORDER BY created_at_ms ASC, event_id ASC LIMIT ? FOR UPDATE SKIP LOCKED + `, + args: []any{appCode, nowMS}, + }, + { + query: ` + SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms + FROM user_outbox FORCE INDEX (idx_user_outbox_status_created) + WHERE app_code = ? AND status = 'pending' AND next_retry_at_ms <= ? AND lock_until_ms = 0 + ORDER BY created_at_ms ASC, event_id ASC + LIMIT ? FOR UPDATE SKIP LOCKED `, args: []any{appCode, nowMS}, }, diff --git a/services/user-service/internal/storage/mysql/user_outbox_claim_priority_test.go b/services/user-service/internal/storage/mysql/user_outbox_claim_priority_test.go new file mode 100644 index 00000000..53b7eed0 --- /dev/null +++ b/services/user-service/internal/storage/mysql/user_outbox_claim_priority_test.go @@ -0,0 +1,79 @@ +package mysql + +import ( + "context" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "hyapp/pkg/appcode" +) + +func TestClaimPendingUserOutboxPrioritizesRecoveryOverPendingAtBatchOne(t *testing.T) { + tests := []struct { + name string + eventID string + recoveryStatus string + expectEmptyRetry bool + }{ + {name: "due retry", eventID: "user-retry", recoveryStatus: "retryable"}, + {name: "expired run", eventID: "user-expired", recoveryStatus: "running", expectEmptyRetry: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("create sqlmock: %v", err) + } + defer db.Close() + + mock.ExpectBegin() + if tt.expectEmptyRetry { + mock.ExpectQuery(userOutboxRetryClaimPattern()). + WithArgs("lalu", sqlmock.AnyArg(), 1). + WillReturnRows(userOutboxClaimRows()) + } + claimPattern := userOutboxRetryClaimPattern() + if tt.recoveryStatus == "running" { + claimPattern = userOutboxExpiredClaimPattern() + } + mock.ExpectQuery(claimPattern). + WithArgs("lalu", sqlmock.AnyArg(), 1). + WillReturnRows(userOutboxClaimRows().AddRow( + "lalu", tt.eventID, "UserRegistered", "user", int64(1001), `{}`, int64(200), + )) + mock.ExpectExec(`(?s)UPDATE user_outbox\s+SET status = 'running', worker_id = \?, lock_until_ms = \?, updated_at_ms = \?\s+WHERE app_code = \? AND event_id = \?`). + WithArgs("worker-1", sqlmock.AnyArg(), sqlmock.AnyArg(), "lalu", tt.eventID). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + + records, err := (&Repository{db: db}).ClaimPendingUserOutbox( + appcode.WithContext(context.Background(), "lalu"), "worker-1", 1, + ) + if err != nil { + t.Fatalf("claim user recovery row: %v", err) + } + if len(records) != 1 || records[0].EventID != tt.eventID { + t.Fatalf("batch=1 must claim %s before an older pending row: %+v", tt.recoveryStatus, records) + } + // No pending query is registered: issuing one after the recovery row filled + // the batch makes sqlmock fail, protecting the starvation boundary directly. + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("user claim priority mismatch: %v", err) + } + }) + } +} + +func userOutboxClaimRows() *sqlmock.Rows { + return sqlmock.NewRows([]string{ + "app_code", "event_id", "event_type", "aggregate_type", "aggregate_id", "payload_json", "created_at_ms", + }) +} + +func userOutboxRetryClaimPattern() string { + return `(?s)FROM user_outbox FORCE INDEX \(idx_user_outbox_status_created\).*WHERE app_code = \? AND status = 'retryable' AND next_retry_at_ms <= \? AND lock_until_ms = 0.*FOR UPDATE SKIP LOCKED` +} + +func userOutboxExpiredClaimPattern() string { + return `(?s)FROM user_outbox FORCE INDEX \(idx_user_outbox_lock\).*WHERE app_code = \? AND status = 'running' AND lock_until_ms <= \?.*FOR UPDATE SKIP LOCKED` +} diff --git a/services/wallet-service/internal/storage/mysql/outbox.go b/services/wallet-service/internal/storage/mysql/outbox.go index 2bdfe80b..79c0cd11 100644 --- a/services/wallet-service/internal/storage/mysql/outbox.go +++ b/services/wallet-service/internal/storage/mysql/outbox.go @@ -142,16 +142,14 @@ func (r *Repository) ClaimPendingWalletOutboxFiltered(ctx context.Context, worke defer func() { _ = tx.Rollback() }() records := make([]WalletOutboxRecord, 0, limit) + // Recovery work must claim before fresh pending facts. A continuously growing + // pending backlog can otherwise fill every batch and permanently strand both + // due retries and deliveries whose worker lease expired. Keep the states in + // separate SKIP LOCKED queries so each branch retains its status-leading index. claimBranches := []struct { query string args []any }{ - { - query: walletOutboxClaimQuery(pendingIndex, ` - AND next_retry_at_ms IS NULL - AND lock_until_ms IS NULL`, eventFilterSQL), - args: append([]any{appCode, outboxStatusPending}, eventFilterArgs...), - }, { query: walletOutboxClaimQuery(pendingIndex, ` AND next_retry_at_ms <= ? @@ -160,9 +158,15 @@ func (r *Repository) ClaimPendingWalletOutboxFiltered(ctx context.Context, worke }, { query: walletOutboxClaimQuery(claimIndex, ` - AND lock_until_ms <= ?`, eventFilterSQL), + AND lock_until_ms <= ?`, eventFilterSQL), args: append([]any{appCode, outboxStatusDelivering, nowMS}, eventFilterArgs...), }, + { + query: walletOutboxClaimQuery(pendingIndex, ` + AND next_retry_at_ms IS NULL + AND lock_until_ms IS NULL`, eventFilterSQL), + args: append([]any{appCode, outboxStatusPending}, eventFilterArgs...), + }, } for _, branch := range claimBranches { remaining := limit - len(records) diff --git a/services/wallet-service/internal/storage/mysql/outbox_claim_priority_test.go b/services/wallet-service/internal/storage/mysql/outbox_claim_priority_test.go new file mode 100644 index 00000000..a23de4ff --- /dev/null +++ b/services/wallet-service/internal/storage/mysql/outbox_claim_priority_test.go @@ -0,0 +1,83 @@ +package mysql + +import ( + "context" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "hyapp/pkg/appcode" +) + +func TestClaimPendingWalletOutboxPrioritizesRecoveryOverPendingAtBatchOne(t *testing.T) { + tests := []struct { + name string + eventID string + recoveryStatus string + expectEmptyRetry bool + }{ + {name: "due retry", eventID: "wallet-retry", recoveryStatus: outboxStatusRetryable}, + {name: "expired delivery", eventID: "wallet-expired", recoveryStatus: outboxStatusDelivering, expectEmptyRetry: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("create sqlmock: %v", err) + } + defer db.Close() + + mock.ExpectBegin() + if tt.expectEmptyRetry { + mock.ExpectQuery(walletOutboxRetryClaimPattern()). + WithArgs("lalu", outboxStatusRetryable, sqlmock.AnyArg(), 1). + WillReturnRows(walletOutboxClaimRows()) + } + claimPattern := walletOutboxRetryClaimPattern() + if tt.recoveryStatus == outboxStatusDelivering { + claimPattern = walletOutboxExpiredClaimPattern() + } + mock.ExpectQuery(claimPattern). + WithArgs("lalu", tt.recoveryStatus, sqlmock.AnyArg(), 1). + WillReturnRows(walletOutboxClaimRows().AddRow( + "lalu", tt.eventID, "WalletBalanceChanged", "tx-1", "cmd-1", int64(1001), + "COIN", int64(1), int64(0), `{}`, int64(200), 0, "", "old-worker", int64(100), + )) + lockUntilMS := time.Now().UTC().Add(time.Minute).UnixMilli() + mock.ExpectExec(`(?s)UPDATE wallet_outbox\s+SET status = \?, worker_id = \?, lock_until_ms = \?, updated_at_ms = \?\s+WHERE app_code = \? AND event_id = \?`). + WithArgs(outboxStatusDelivering, "worker-1", lockUntilMS, sqlmock.AnyArg(), "lalu", tt.eventID). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + + records, err := (&Repository{db: db}).ClaimPendingWalletOutbox( + appcode.WithContext(context.Background(), "lalu"), "worker-1", 1, lockUntilMS, + ) + if err != nil { + t.Fatalf("claim wallet recovery row: %v", err) + } + if len(records) != 1 || records[0].EventID != tt.eventID { + t.Fatalf("batch=1 must claim %s before an older pending row: %+v", tt.recoveryStatus, records) + } + // No pending query is registered: issuing one after the recovery row filled + // the batch makes sqlmock fail, protecting the starvation boundary directly. + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("wallet claim priority mismatch: %v", err) + } + }) + } +} + +func walletOutboxClaimRows() *sqlmock.Rows { + return sqlmock.NewRows([]string{ + "app_code", "event_id", "event_type", "transaction_id", "command_id", "user_id", "asset_type", + "available_delta", "frozen_delta", "payload", "created_at_ms", "retry_count", "last_error", "worker_id", "lock_until_ms", + }) +} + +func walletOutboxRetryClaimPattern() string { + return `(?s)FROM wallet_outbox FORCE INDEX \(idx_wallet_outbox_pending\).*WHERE app_code = \? AND status = \?\s+AND next_retry_at_ms <= \?\s+AND lock_until_ms IS NULL.*FOR UPDATE SKIP LOCKED` +} + +func walletOutboxExpiredClaimPattern() string { + return `(?s)FROM wallet_outbox FORCE INDEX \(idx_wallet_outbox_claim\).*WHERE app_code = \? AND status = \?\s+AND lock_until_ms <= \?.*FOR UPDATE SKIP LOCKED` +}