diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 92585af6..e0a88c22 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1226,8 +1226,8 @@ the local/CI/community transport, not a silent production fallback. | # | Task | Acceptance | |---|---|---| -| 8.39 `[D:8.3,8.14,8.17]` | **IN PROGRESS.** `MatchmakingState` now projects queue → proposal → allocation/process-ready/assignment-ready/connect/live plus terminal failure states; autoload `ControlPlaneClient` provides authenticated queue create/recovery/heartbeat/cancel and proposal response requests with idempotency/revision headers; `matchmaking.tscn`/`matchmaking.gd` expose the state and authoritative actions from the main menu; authenticated proposal recovery now reconciles missed proposal events and expires them at read time | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd` and `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary` reject stale/gapped/conflicting updates, validate endpoint/token/payload normalization, preserve idempotent duplicates, and guarantee visible phase/terminal copy; server-pushed allocation events, wait/latency explanations and Godot runtime verification remain | -| 8.40 `[D:8.3,8.14]` | **IN PROGRESS.** Pure Go revisioned replica reducer rejects gaps for REST resync, makes duplicate/out-of-order events idempotent, and resumes from the authoritative snapshot revision; authenticated queue-ticket recovery now has an owner-checked REST read; Godot client projection persists non-secret ticket/proposal state, forces authoritative recovery after restart, and can replay a lost queue-create response with the original ticket/idempotency key | `server/domain/sync.go`, `server/api/service.go`, `service_test.go`, `matchmaking_state.gd` and `control_plane_client.gd` cover gap, snapshot, replay, same-revision conflict, owner-only ticket recovery, expired-ticket terminal handling, malformed restart snapshots and API-level duplicate-create replay/conflict; authenticated WebSocket transport and live Godot verification remain | +| 8.39 `[D:8.3,8.14,8.17]` | **IN PROGRESS.** `MatchmakingState` now projects queue → proposal → allocation/process-ready/assignment-ready/connect/live plus terminal failure states; autoload `ControlPlaneClient` provides authenticated queue create/recovery/heartbeat/cancel and proposal response requests with idempotency/revision headers; `matchmaking.tscn`/`matchmaking.gd` expose the state and authoritative actions from the main menu; authenticated proposal recovery now reconciles missed proposal events and expires them at read time; the Go API now provides targeted authenticated revisioned event publication | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd`, `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary` and `TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents` reject stale/gapped/conflicting updates, validate endpoint/token/payload normalization, preserve idempotent duplicates, and guarantee visible phase/terminal copy; wiring durable allocation events into the outbox, wait/latency explanations and Godot runtime verification remain | +| 8.40 `[D:8.3,8.14]` | **IN PROGRESS.** Pure Go revisioned replica reducer rejects gaps for REST resync, makes duplicate/out-of-order events idempotent, and resumes from the authoritative snapshot revision; authenticated queue-ticket recovery now has an owner-checked REST read; Godot client projection persists non-secret ticket/proposal state, forces authoritative recovery after restart, and can replay a lost queue-create response with the original ticket/idempotency key; the Go API now exposes an authenticated `/v1/events` WebSocket with bounded per-player queues, strict upgrade validation and REST-resync-safe slow-client failure | `server/domain/sync.go`, `server/api/events.go`, `server/api/service.go`, `service_test.go`, `matchmaking_state.gd` and `control_plane_client.gd` cover gap, snapshot, replay, same-revision conflict, owner-only ticket recovery, expired-ticket terminal handling, malformed restart snapshots, API-level duplicate-create replay/conflict, authenticated handshake/key rejection and targeted event delivery; Godot WebSocket client wiring, durable outbox fan-out, reconnect/resync orchestration and live Godot verification remain | | 8.41 `[D:7.8,8.9,8.31,8.40]` | **IN PROGRESS.** Authenticated `GET /v1/assignments/{matchId}` now exposes only a validated, player-scoped assignment view; Godot `AssignmentState`/`ControlPlaneClient.fetch_assignment()` bind the response to the authenticated player, recheck expiry, preserve explicit transport/slot/join authorisation and do not connect before assignment-ready | `server/api/service.go`, `service_test.go`, `assignment_state.gd` and `test_assignment_state.gd` cover participant/identity/expiry/shape/transport boundaries and assignment recovery; signed manifest-to-player persistence, SDR relay-ticket installation, `hello` join-authorisation wiring, fencing integration and live Godot verification remain | | 8.42 `[D:8.22,8.23,8.24,8.40]` | **IN PROGRESS.** `RankedProfileState` and `ControlPlaneClient.fetch_ranked_profile()` expose the backend-authoritative rating/RD/volatility/games/tier/provisional/season view; matchmaking UI displays provisional/tier status without client-side rating math | `test_control_plane_client.gd` validates profile shape, numeric safety and provisional display; ranked profile fetch/display, committed revision after reconnect, abandon status and season countdown remain dependent on live auth/backend events and Godot runtime verification | | 8.43 `[D:8.39,8.40,8.41]` | **IN PROGRESS.** Matchmaking client now distinguishes expired queue recovery, session expiry, missing records and retryable control-plane outages; a 401 clears the in-memory token, emits `session_expired` and disables retry until a new session is configured; terminal messages remain visible and active searches are not falsely failed on transient errors | `MatchmakingState` and `ControlPlaneClient` tests cover explicit expiry and the existing terminal/retry-safe state paths; decline, version mismatch, regional outage retry UI, failed reconnect, duplicate-action recovery and live Godot verification remain | diff --git a/server/api/events.go b/server/api/events.go new file mode 100644 index 00000000..6d4fdc85 --- /dev/null +++ b/server/api/events.go @@ -0,0 +1,276 @@ +package api + +import ( + "bufio" + "crypto/sha1" + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "time" +) + +const ( + webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + maxWebSocketFrame = 64 << 10 + eventQueueCapacity = 32 +) + +// ControlPlaneEvent is the server-to-client envelope defined by the v1 +// WebSocket contract. PlayerID is routing metadata and is never serialized. +type ControlPlaneEvent struct { + Event string `json:"event"` + Revision uint64 `json:"revision"` + ResourceID string `json:"resource_id"` + OccurredAt time.Time `json:"occurred_at"` + State string `json:"state,omitempty"` + Code string `json:"code,omitempty"` + MatchID string `json:"match_id,omitempty"` + ServerID string `json:"server_id,omitempty"` + PlayerID string `json:"-"` +} + +type eventSubscriber struct { + playerID string + queue chan []byte +} + +type eventHub struct { + mu sync.Mutex + subscribers map[*eventSubscriber]struct{} +} + +func newEventHub() *eventHub { + return &eventHub{subscribers: make(map[*eventSubscriber]struct{})} +} + +func (h *eventHub) subscribe(playerID string) *eventSubscriber { + subscriber := &eventSubscriber{playerID: playerID, queue: make(chan []byte, eventQueueCapacity)} + h.mu.Lock() + h.subscribers[subscriber] = struct{}{} + h.mu.Unlock() + return subscriber +} + +func (h *eventHub) unsubscribe(subscriber *eventSubscriber) { + h.mu.Lock() + delete(h.subscribers, subscriber) + close(subscriber.queue) + h.mu.Unlock() +} + +func (h *eventHub) publish(event ControlPlaneEvent) error { + if event.PlayerID == "" || event.Event == "" || event.ResourceID == "" || event.OccurredAt.IsZero() { + return errors.New("invalid control-plane event") + } + payload, err := json.Marshal(event) + if err != nil { + return err + } + h.mu.Lock() + defer h.mu.Unlock() + for subscriber := range h.subscribers { + if subscriber.playerID != event.PlayerID { + continue + } + select { + case subscriber.queue <- payload: + default: + // A slow client must not block state publication for other clients. + // Closing its queue makes the connection fail closed and recover via + // REST resync rather than silently dropping an unbounded history. + delete(h.subscribers, subscriber) + close(subscriber.queue) + } + } + return nil +} + +func (s *Service) controlPlaneEvent(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method_not_allowed") + return + } + if !isWebSocketUpgrade(r) || !validWebSocketKey(r.Header.Get("Sec-WebSocket-Key")) { + writeError(w, http.StatusBadRequest, "invalid_websocket_upgrade") + return + } + playerID, ok := s.authenticate(w, r) + if !ok { + return + } + hijacker, ok := w.(http.Hijacker) + if !ok { + writeError(w, http.StatusNotImplemented, "websocket_unavailable") + return + } + connection, buffered, err := hijacker.Hijack() + if err != nil { + return + } + defer connection.Close() + accept := websocketAccept(r.Header.Get("Sec-WebSocket-Key")) + if _, err := fmt.Fprintf(buffered, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: %s\r\n\r\n", accept); err != nil { + return + } + if err := buffered.Flush(); err != nil { + return + } + subscriber := s.getEventHub().subscribe(playerID) + defer s.getEventHub().unsubscribe(subscriber) + + var writeMu sync.Mutex + done := make(chan struct{}) + go func() { + defer close(done) + readWebSocketFrames(connection, &writeMu) + }() + for { + select { + case payload, open := <-subscriber.queue: + if !open { + return + } + writeMu.Lock() + err := writeWebSocketFrame(connection, 0x1, payload) + writeMu.Unlock() + if err != nil { + return + } + case <-done: + return + } + } +} + +func (s *Service) getEventHub() *eventHub { + s.eventsMu.Lock() + defer s.eventsMu.Unlock() + if s.events == nil { + s.events = newEventHub() + } + return s.events +} + +// PublishControlPlaneEvent routes an already-authorized event to the matching +// authenticated player connection. Durable callers should publish from their +// outbox after commit; this in-memory hub is deliberately non-authoritative. +func (s *Service) PublishControlPlaneEvent(event ControlPlaneEvent) error { + return s.getEventHub().publish(event) +} + +func isWebSocketUpgrade(r *http.Request) bool { + return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") && headerContainsToken(r.Header.Values("Connection"), "upgrade") +} + +func headerContainsToken(values []string, wanted string) bool { + for _, value := range values { + for _, token := range strings.Split(value, ",") { + if strings.EqualFold(strings.TrimSpace(token), wanted) { + return true + } + } + } + return false +} + +func websocketAccept(key string) string { + digest := sha1.Sum([]byte(key + webSocketGUID)) + return base64.StdEncoding.EncodeToString(digest[:]) +} + +func validWebSocketKey(key string) bool { + decoded, err := base64.StdEncoding.DecodeString(key) + return err == nil && len(decoded) == 16 +} + +func readWebSocketFrames(connection net.Conn, writeMu *sync.Mutex) { + reader := bufio.NewReader(connection) + for { + opcode, _, err := readWebSocketFrame(reader) + if err != nil || opcode == 0x8 { + return + } + if opcode == 0x9 { + writeMu.Lock() + _ = writeWebSocketFrame(connection, 0xA, nil) + writeMu.Unlock() + } + } +} + +func readWebSocketFrame(reader *bufio.Reader) (byte, []byte, error) { + first, err := reader.ReadByte() + if err != nil { + return 0, nil, err + } + second, err := reader.ReadByte() + if err != nil { + return 0, nil, err + } + if first&0x70 != 0 || first&0x80 == 0 { + return 0, nil, errors.New("unsupported websocket frame") + } + if second&0x80 == 0 { + return 0, nil, errors.New("unmasked websocket frame") + } + length := int64(second & 0x7f) + if length == 126 { + var extended uint16 + if err := binary.Read(reader, binary.BigEndian, &extended); err != nil { + return 0, nil, err + } + length = int64(extended) + } else if length == 127 { + var extended uint64 + if err := binary.Read(reader, binary.BigEndian, &extended); err != nil || extended > maxWebSocketFrame { + return 0, nil, errors.New("websocket frame too large") + } + length = int64(extended) + } + if length > maxWebSocketFrame { + return 0, nil, errors.New("websocket frame too large") + } + var mask [4]byte + if _, err := io.ReadFull(reader, mask[:]); err != nil { + return 0, nil, err + } + payload := make([]byte, length) + if _, err := io.ReadFull(reader, payload); err != nil { + return 0, nil, err + } + for i := range payload { + payload[i] ^= mask[i%4] + } + return first & 0x0f, payload, nil +} + +func writeWebSocketFrame(connection net.Conn, opcode byte, payload []byte) error { + if len(payload) > maxWebSocketFrame { + return errors.New("websocket frame too large") + } + header := []byte{0x80 | opcode} + switch { + case len(payload) < 126: + header = append(header, byte(len(payload))) + case len(payload) <= 65535: + header = append(header, 126, 0, 0) + binary.BigEndian.PutUint16(header[len(header)-2:], uint16(len(payload))) + default: + header = append(header, 127) + var extended [8]byte + binary.BigEndian.PutUint64(extended[:], uint64(len(payload))) + header = append(header, extended[:]...) + } + if _, err := connection.Write(header); err != nil { + return err + } + _, err := connection.Write(payload) + return err +} diff --git a/server/api/service.go b/server/api/service.go index ec001e16..8bc401f4 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -72,6 +72,8 @@ type Service struct { RankedProfiles map[string]domain.RankedProfile TierPolicy domain.TierPolicy proposalMu sync.Mutex + eventsMu sync.Mutex + events *eventHub } func (s *Service) Handler() http.Handler { @@ -84,6 +86,7 @@ func (s *Service) Handler() http.Handler { mux.HandleFunc("/v1/assignments/", s.assignment) mux.HandleFunc("/v1/profile/ranked", s.rankedProfile) mux.HandleFunc("/v1/probes/", s.probe) + mux.HandleFunc("/v1/events", s.controlPlaneEvent) // The public contract is served below /api/v1. Keep the original /v1 // routes for the Godot client while exposing the documented names. mux.HandleFunc("/api/v1/session/steam", s.steamSession) @@ -92,6 +95,7 @@ func (s *Service) Handler() http.Handler { mux.HandleFunc("/api/v1/queue/tickets/", s.contractQueueMutation) mux.HandleFunc("/api/v1/proposals/", s.contractProposalMutation) mux.HandleFunc("/api/v1/assignments/", s.contractAssignment) + mux.HandleFunc("/api/v1/events", s.controlPlaneEvent) return mux } diff --git a/server/api/service_test.go b/server/api/service_test.go index fa4da1ac..028499b9 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -1,8 +1,13 @@ package api import ( + "bufio" "context" + "encoding/binary" "encoding/json" + "errors" + "io" + "net" "net/http" "net/http/httptest" "strings" @@ -161,6 +166,97 @@ func TestDocumentedContractRoutesAdaptToServiceAPI(t *testing.T) { } } +func TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents(t *testing.T) { + service := &Service{SessionBackend: &sessionBackendSpy{}} + server := httptest.NewServer(service.Handler()) + defer server.Close() + invalid, err := http.NewRequest(http.MethodGet, server.URL+"/v1/events", nil) + if err != nil { + t.Fatal(err) + } + invalid.Header.Set("Upgrade", "websocket") + invalid.Header.Set("Connection", "Upgrade") + invalid.Header.Set("Sec-WebSocket-Key", "not-a-websocket-key") + invalid.Header.Set("Authorization", "Bearer session-1:token-1") + invalidResponse, err := server.Client().Do(invalid) + if err != nil { + t.Fatal(err) + } + _ = invalidResponse.Body.Close() + if invalidResponse.StatusCode != http.StatusBadRequest { + t.Fatalf("invalid handshake status = %d", invalidResponse.StatusCode) + } + connection, err := net.Dial("tcp", strings.TrimPrefix(server.URL, "http://")) + if err != nil { + t.Fatal(err) + } + defer connection.Close() + _, err = io.WriteString(connection, "GET /v1/events HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nAuthorization: Bearer session-1:token-1\r\n\r\n") + if err != nil { + t.Fatal(err) + } + reader := bufio.NewReader(connection) + status, err := reader.ReadString('\n') + if err != nil { + t.Fatal(err) + } + if !strings.Contains(status, "101 Switching Protocols") { + t.Fatalf("handshake status = %q", status) + } + for { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatal(err) + } + if line == "\r\n" { + break + } + } + time.Sleep(10 * time.Millisecond) + if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: 1, ResourceID: "ticket-1234567890123456", OccurredAt: time.Unix(1000, 0).UTC(), State: "QUEUED", PlayerID: "player-1"}); err != nil { + t.Fatal(err) + } + if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: 2, ResourceID: "ticket-1234567890123456", OccurredAt: time.Unix(1001, 0).UTC(), State: "PROPOSED", PlayerID: "player-2"}); err != nil { + t.Fatal(err) + } + first, err := readServerWebSocketFrame(reader) + if err != nil { + t.Fatal(err) + } + var event ControlPlaneEvent + if err := json.Unmarshal(first, &event); err != nil { + t.Fatal(err) + } + if event.PlayerID != "" || event.Revision != 1 || event.ResourceID != "ticket-1234567890123456" || event.State != "QUEUED" { + t.Fatalf("event = %+v", event) + } +} + +func readServerWebSocketFrame(reader *bufio.Reader) ([]byte, error) { + first, err := reader.ReadByte() + if err != nil { + return nil, err + } + second, err := reader.ReadByte() + if err != nil { + return nil, err + } + if first&0x0f != 0x1 || second&0x80 != 0 { + return nil, errors.New("unexpected server websocket frame") + } + length := int(second & 0x7f) + if length == 126 { + var extended uint16 + if err := binary.Read(reader, binary.BigEndian, &extended); err != nil { + return nil, err + } + length = int(extended) + } + payload := make([]byte, length) + _, err = io.ReadFull(reader, payload) + return payload, err +} + func TestQueueAPIRejectsUnauthenticatedUnknownAndOversizedInput(t *testing.T) { service := &Service{Sessions: domain.NewSessionStore(), Queue: domain.NewQueue(), Candidate: func(string, string) (domain.Candidate, error) { return domain.Candidate{}, nil }} server := httptest.NewServer(service.Handler())