test(multiplayer): verify two-player proposal round trip

server/cmd/matcher/main.go

@@ -112,6 +112,7 @@ func main() {
 			}
 			return domain.PrepareProposal(id, playlist, formation, nil, domain.RankedArena{}, at)
 		},
+		OnError: func(err error) { log.Printf("matcher pass: %v", err) },
 	}
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()

server/cmd/testkit-api/main.go

@@ -16,6 +16,7 @@ import (
 	"crypto/sha256"
 	"database/sql"
 	"encoding/hex"
+	"encoding/json"
 	"flag"
 	"fmt"
 	"net"
@@ -54,7 +55,7 @@ func main() {
 	if err := migrations.Apply(startupCtx, db, *migrationDir); err != nil {
 		fatalf("apply migrations: %v", err)
 	}
-	handler := (&api.Service{
+	service := &api.Service{
 		SessionBackend:        store.PostgresSessions{DB: db},
 		SessionIssuer:         store.PostgresSessions{DB: db},
 		SteamLogin:            fakeSteamLogin{db: db},
@@ -68,7 +69,8 @@ func main() {
 		ProbeRecorder:         store.PostgresQueue{DB: db},
 		WorkloadVerify:        api.WorkloadVerifierFromSignedToken([]byte(*workloadSecret), db),
 		Now:                   func() time.Time { return time.Now().UTC() },
-	}).Handler()
+	}
+	handler := service.Handler()
 	listener, err := net.Listen("tcp", *listen)
 	if err != nil {
 		fatalf("listen: %v", err)
@@ -79,6 +81,7 @@ func main() {
 	go func() { serveErr <- server.Serve(listener) }()
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
+	go dispatchProposalOutbox(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {
@@ -91,6 +94,46 @@ func main() {
 	}
 }
 
+func dispatchProposalOutbox(ctx context.Context, db *sql.DB, service *api.Service) {
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			events, err := store.ReadUnpublishedOutbox(ctx, db, 100)
+			if err != nil {
+				continue
+			}
+			for _, event := range events {
+				if event.EventType != "proposal_changed" {
+					continue
+				}
+				var controlEvent api.ControlPlaneEvent
+				var envelope struct {
+					Event      string    `json:"event"`
+					Revision   uint64    `json:"revision"`
+					ResourceID string    `json:"resource_id"`
+					OccurredAt time.Time `json:"occurred_at"`
+					State      string    `json:"state"`
+					PlayerIDs  []string  `json:"player_ids"`
+				}
+				if err := json.Unmarshal(event.Payload, &envelope); err != nil {
+					continue
+				}
+				for _, playerID := range envelope.PlayerIDs {
+					controlEvent = api.ControlPlaneEvent{Event: envelope.Event, Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, PlayerID: playerID}
+					if err := service.PublishControlPlaneEvent(controlEvent); err != nil {
+						continue
+					}
+				}
+				_ = store.MarkOutboxPublished(ctx, db, event.EventID, time.Now().UTC())
+			}
+		}
+	}
+}
+
 // fakeSteamLogin derives a deterministic identity from the ticket string
 // itself (never a real Steam Web API ticket in this binary) and ensures its
 // identities row exists so session issuance's foreign key is satisfied.

server/matcher/worker.go

@@ -47,6 +47,7 @@ type Worker struct {
 	Now      func() time.Time
 	NextID   func() string
 	Prepare  PrepareFunc
+	OnError  func(error)
 }
 
 // Run polls until cancellation. A failed attempt is returned so a supervisor
@@ -57,6 +58,9 @@ func (w Worker) Run(ctx context.Context, interval time.Duration) error {
 	}
 	for {
 		if _, err := w.RunOnce(ctx); err != nil {
+			if w.OnError != nil {
+				w.OnError(err)
+			}
 			if errors.Is(err, ErrWorkerNotConfigured) || errors.Is(err, ErrUnsupportedPlaylist) || errors.Is(err, ErrInvalidMatcherSize) {
 				return err
 			}

server/store/proposal_sql.go

@@ -3,6 +3,7 @@ package store
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -13,6 +14,10 @@ const ProposalInsertSQL = `INSERT INTO proposals
     (proposal_id, playlist, state, expires_at, revision, match_region, match_protocol)
 VALUES ($1, $2, 'OPEN', $3, 0, NULLIF($4, ''), NULLIF($5, 0))`
 
+const ProposalOutboxInsertSQL = `INSERT INTO outbox
+    (event_id, aggregate_type, aggregate_id, revision, event_type, payload)
+VALUES ($1, 'proposal', $2, 0, 'proposal_changed', $3)`
+
 // CreateProposal atomically claims the queue tickets and creates the proposal.
 // Every statement runs inside the same SERIALIZABLE retry callback; callers
 // must never publish a proposal from a cache-only candidate list.
@@ -27,6 +32,7 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t
 		if _, err := tx.ExecContext(ctx, ProposalInsertSQL, proposal.ProposalID, proposal.Playlist, proposal.ExpiresAt, proposal.Region, proposal.Protocol); err != nil {
 			return err
 		}
+		players := make([]string, 0, len(proposal.Participants))
 		for _, participant := range proposal.Participants {
 			ticketID := ticketIDs[participant.PlayerID]
 			if participant.PlayerID == "" || ticketID == "" {
@@ -46,6 +52,17 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t
 			if changed != 1 {
 				return fmt.Errorf("queue ticket claim lost")
 			}
+			players = append(players, participant.PlayerID)
+		}
+		payload, err := json.Marshal(map[string]any{
+			"event": "proposal_changed", "revision": uint64(0), "resource_id": proposal.ProposalID,
+			"occurred_at": now, "state": string(proposal.State), "player_ids": players,
+		})
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, ProposalOutboxInsertSQL, proposal.ProposalID, proposal.ProposalID, payload); err != nil {
+			return err
 		}
 		return nil
 	})