diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 7d9696e2..11b6bff1 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1202,7 +1202,7 @@ the local/CI/community transport, not a silent production fallback. | 8.22 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked profile exposes the first ten games as provisional, derives tiers only through validated backend-owned rating bands, and keeps casual ratings outside the API; authenticated HTTP now returns the authoritative ranked view | `server/domain/rating.go`, `tier_test.go` and `server/api/service.go` cover provisional override, exact band boundaries, malformed policy rejection, session authentication and ranked-only response fields; persisted tier policy, client UI and reconnect transport remain | | 8.23 `[D:8.21]` | **IN PROGRESS.** Pure Go ranked-only season policy compresses 25% toward 1500, clamps RD to 200–350, preserves volatility/history, is idempotent by season ID, and defines exact 12-week windows/due detection; migration and Go store now persist a per-player/per-season marker and rating update atomically | `server/domain/rating.go`, `season_test.go`, `server/migrations/0001_initial.sql` and `server/store/season_sql.go` cover compression, floor/cap, duplicate replay, window boundary, completed-season idempotence, row locking and conflict-safe rollover markers; live PostgreSQL execution and maintenance scheduler remain | | 8.24 `[D:8.9,8.20,8.21]` | **IN PROGRESS.** Pure Go ranked connection policy binds match/server/player/team/slot/protocol, supports 60 s reclaim with server-owned generations, fences old connections, and applies the rolling 7-day 5 m/15 m/1 h/24 h abandon ladder; canonical signed authorisation issuance/verification now gates admission | `server/domain/reconnect.go`, `join_auth.go` and adversarial fixtures cover repeated backend-independent reclaim, all signed claim binding, tampering, failed verification, old-generation fencing, grace boundary and deterministic cooldown audit ordering; persistent lease fencing, join transport and full match/result integration remain | -| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Pure Go result policy binds match/server/workload identity, hashes canonical payloads, makes identical retries idempotent, leaves conflicts inert, separates integrity eligibility, classifies roster/simulation/result/fairness evidence, validates annotation signatures/digests, and exposes 5 m alert/30 m review delivery thresholds; Go store SQL now executes the receipt → match lock → completion → receipt acknowledgment → outbox boundary atomically | `server/domain/result.go`, `workload.go` and `server/store/result_sql.go` plus adversarial fixtures cover credential binding, duplicate/conflict, annotation forgery, delivery-outage-versus-integrity classification, commit ordering and idempotent SQL reconciliation; production credential verification, Agones annotation persistence/reconciliation, rating-lock integration, live PostgreSQL execution and integrity evidence adapters remain | +| 8.25 `[D:8.10,8.24]` | **IN PROGRESS.** Pure Go result policy binds match/server/workload identity, hashes canonical payloads, makes identical retries idempotent, leaves conflicts inert, separates integrity eligibility, classifies roster/simulation/result/fairness evidence, validates annotation signatures/digests, and exposes 5 m alert/30 m review delivery thresholds; Go store SQL now executes the receipt → match lock → completion → receipt acknowledgment → outbox boundary atomically, and exposes bounded ordered outbox reads plus publish acknowledgements for replayable fan-out | `server/domain/result.go`, `workload.go`, `server/store/result_sql.go` and `outbox.go` plus adversarial fixtures cover credential binding, duplicate/conflict, annotation forgery, delivery-outage-versus-integrity classification, commit ordering, idempotent SQL reconciliation and unpublished-event replay/ack boundaries; production credential verification, Agones annotation persistence/reconciliation, rating-lock integration, live PostgreSQL execution and integrity evidence adapters remain | #### 8D — Agones, allocation and regional scaling diff --git a/server/store/outbox.go b/server/store/outbox.go new file mode 100644 index 00000000..2da0f0b5 --- /dev/null +++ b/server/store/outbox.go @@ -0,0 +1,81 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// OutboxEvent is the durable hand-off between a committed domain mutation and +// transient WebSocket delivery. Consumers must make delivery idempotent by +// event ID and only acknowledge after successful fan-out. +type OutboxEvent struct { + EventID string + AggregateType string + AggregateID string + Revision uint64 + EventType string + Payload []byte + CreatedAt time.Time + PublishedAt *time.Time +} + +const OutboxUnpublishedSelectSQL = `SELECT event_id, aggregate_type, aggregate_id, revision, + event_type, payload, created_at, published_at +FROM outbox +WHERE published_at IS NULL +ORDER BY created_at, event_id +LIMIT $1` + +const OutboxMarkPublishedSQL = `UPDATE outbox +SET published_at = $2 +WHERE event_id = $1 AND published_at IS NULL` + +var ErrOutboxEventNotFound = fmt.Errorf("outbox event not found or already published") + +// ReadUnpublishedOutbox returns a bounded, stable ordered batch. It does not +// mark rows before delivery: a worker crash therefore leaves events replayable. +func ReadUnpublishedOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) { + if db == nil || limit < 1 || limit > 1000 { + return nil, fmt.Errorf("invalid outbox read arguments") + } + rows, err := db.QueryContext(ctx, OutboxUnpublishedSelectSQL, limit) + if err != nil { + return nil, err + } + defer rows.Close() + events := make([]OutboxEvent, 0, limit) + for rows.Next() { + var event OutboxEvent + if err := rows.Scan(&event.EventID, &event.AggregateType, &event.AggregateID, &event.Revision, &event.EventType, &event.Payload, &event.CreatedAt, &event.PublishedAt); err != nil { + return nil, err + } + events = append(events, event) + } + if err := rows.Err(); err != nil { + return nil, err + } + return events, nil +} + +// MarkOutboxPublished acknowledges one event only if it is still unpublished. +// Repeated acknowledgement is reported to the caller so a worker cannot +// mistake an already-completed delivery for a fresh one. +func MarkOutboxPublished(ctx context.Context, db *sql.DB, eventID string, publishedAt time.Time) error { + if db == nil || eventID == "" || publishedAt.IsZero() { + return fmt.Errorf("invalid outbox acknowledgement arguments") + } + result, err := db.ExecContext(ctx, OutboxMarkPublishedSQL, eventID, publishedAt) + if err != nil { + return err + } + changed, err := result.RowsAffected() + if err != nil { + return err + } + if changed != 1 { + return ErrOutboxEventNotFound + } + return nil +} diff --git a/server/store/outbox_test.go b/server/store/outbox_test.go new file mode 100644 index 00000000..ea71b80b --- /dev/null +++ b/server/store/outbox_test.go @@ -0,0 +1,34 @@ +package store + +import ( + "testing" + "time" +) + +func TestOutboxSQLPreservesReplayableOrderedReadAndPublishAck(t *testing.T) { + for query, fragments := range map[string][]string{ + OutboxUnpublishedSelectSQL: {"published_at IS NULL", "ORDER BY created_at, event_id", "LIMIT $1"}, + OutboxMarkPublishedSQL: {"published_at = $2", "event_id = $1", "published_at IS NULL"}, + } { + for _, fragment := range fragments { + if !contains(query, fragment) { + t.Fatalf("query %q missing %q", query, fragment) + } + } + } +} + +func TestOutboxAdaptersRejectUnsafeArgumentsWithoutDatabase(t *testing.T) { + if _, err := ReadUnpublishedOutbox(nil, nil, 1); err == nil { + t.Fatal("nil database accepted") + } + if _, err := ReadUnpublishedOutbox(nil, nil, 1001); err == nil { + t.Fatal("unbounded outbox batch accepted") + } + if err := MarkOutboxPublished(nil, nil, "event-1", time.Unix(1000, 0)); err == nil { + t.Fatal("nil database acknowledgement accepted") + } + if err := MarkOutboxPublished(nil, nil, "", time.Unix(1000, 0)); err == nil { + t.Fatal("empty event acknowledgement accepted") + } +}