diff --git a/README.md b/README.md index d88eb61..4f030e9 100644 --- a/README.md +++ b/README.md @@ -56,9 +56,10 @@ The platform provides real-time room-based chat with live analytics: message thr - **Per-connection rate limiting** — token-bucket throttle on inbound messages. - **Token auth** — optional HMAC-signed bearer tokens (enabled when `AUTH_SECRET` is set); falls back to a `userId` query param for local development. - **Configurable CORS / WebSocket origin allowlist.** -- **Graceful degradation** — runs without DynamoDB (chat + live analytics still work, no persistence). +- **Graceful degradation** — runs without DynamoDB (chat + live analytics still work, no persistence). Table init distinguishes an absent table from an unreachable one, so a throttle or transport blip is reported rather than silently retried as a create, and startup waits for the table to become `ACTIVE` before accepting writes. - **Voice Ops Briefings** — an anomaly detector watches the live metrics and narrates traffic events aloud in a control-room register via ElevenLabs TTS. Off by default; see below. - **Polished React frontend** — refined dark theme, design tokens, reusable UI primitives, avatars, virtualized message list, and a live metrics dashboard. +- **Clean shutdown** — the hub's register/unregister/broadcast sends are guarded by its `done` channel, so once the event loop stops, callers return immediately instead of blocking on a channel nothing drains. Without this, every connection's read pump leaks on its deferred unregister. - **Comprehensive tests** — every backend package has a `_test.go`; the suite runs clean under `-race`. ## Quick Start @@ -165,7 +166,9 @@ Recent message history for a room or a user. Optional `?limit=` (default 50, max "timestamp": "2026-06-16T10:30:00Z" } ``` -**Types:** `chat`, `system`, `join`, `leave`. **Validation:** username required (≤50 chars); chat content required (≤1000 chars); `messageId`/`timestamp`/`roomId` are server-authoritative. +**Types:** `chat`, `system`, `join`, `leave`. **Validation:** username required (≤50 chars); chat content required (≤1000 chars). + +`messageId`, `timestamp`, `roomId`, `userId` and `username` are server-authoritative: the server overwrites all five on every inbound frame, regardless of what the client sent. This is load-bearing rather than cosmetic — `messageId` is the DynamoDB sort key and `timestamp` is the GSI sort key, so honouring a client-supplied value would let one connection overwrite another's stored row or pin itself to the head of room history. ## Voice Ops Briefings @@ -186,6 +189,8 @@ An anomaly detector polls the live metrics every 5 seconds and narrates traffic **Detected conditions:** message-rate spike, p99 latency breach, connection surge, traffic dropout — each in a Yellow or Red severity band, debounced to at most one briefing per kind per 5 minutes. +Rate-based conditions are evaluated against **complete** minutes: the newest slot of the 15-minute window is the minute currently in progress, and comparing that partial count against full-minute baselines reads as a dropout at the top of every minute. The detector ignores it. The cost is granularity — a genuine spike or dropout is visible once its minute closes, so up to ~60s later than the 5s poll interval alone would suggest. + A sample script, as spoken: > Ops, analytics. Message-rate spike in the global room — four hundred messages per minute against a baseline of ninety. Monitoring. @@ -284,7 +289,9 @@ npm run build # tsc type-check + vite build npm run lint # eslint ``` -> Note: `TestClient_PingPong` has a documented pre-existing race in the test harness — left as-is intentionally. +> The full suite runs clean under `-race` in a single step, locally and in CI. The +> historical `TestClient_PingPong` race was fixed by making the test's ping counter +> an `atomic.Int64`. ## Deployment diff --git a/backend/pkg/analytics/aggregator.go b/backend/pkg/analytics/aggregator.go index f1fb4dd..545a147 100644 --- a/backend/pkg/analytics/aggregator.go +++ b/backend/pkg/analytics/aggregator.go @@ -19,12 +19,15 @@ func newWindow() *slidingWindow { return &slidingWindow{lastTick: time.Now().Truncate(time.Minute)} } +// increment counts one message into the current minute. The lock is held across +// both the rotation check and the add: releasing it in between lets a concurrent +// rotation advance idx and zero the slot after this call has chosen its index, +// dropping the count into a slot that is immediately cleared. func (w *slidingWindow) increment() { w.mu.Lock() + defer w.mu.Unlock() w.advanceIfNeeded() - idx := w.idx - w.mu.Unlock() - w.slots[idx].Add(1) + w.slots[w.idx].Add(1) } // snapshot returns the counts for the last 15 minutes, oldest first. diff --git a/backend/pkg/analytics/detector.go b/backend/pkg/analytics/detector.go index a02cdfa..5f5e9e5 100644 --- a/backend/pkg/analytics/detector.go +++ b/backend/pkg/analytics/detector.go @@ -179,19 +179,26 @@ func (d *Detector) Check(cur Metrics, now time.Time) []Anomaly { return out } -// rateWindow splits a MessagesPerMinute window into the current minute and the -// mean of the preceding slots. ok is false when the window is too short to -// compare against (fewer than 2 slots). +// rateWindow splits a MessagesPerMinute window into the most recent COMPLETE +// minute and the mean of the minutes before it. The final slot is the minute +// currently in progress: it is a partial count that resets on every rotation, so +// comparing it against full-minute baselines reports a dropout at the top of +// every minute on a busy server. ok is false when the window is too short to hold +// a complete minute plus a baseline. +// +// The cost of ignoring the partial slot is granularity: an event is only visible +// once the minute containing it has closed, so detection lags by up to a minute. func rateWindow(perMin []int64) (current int64, baseline float64, ok bool) { - if len(perMin) < 2 { + if len(perMin) < 3 { return 0, 0, false } - current = perMin[len(perMin)-1] + complete := perMin[:len(perMin)-1] // drop the in-progress slot + current = complete[len(complete)-1] var sum int64 - for _, v := range perMin[:len(perMin)-1] { + for _, v := range complete[:len(complete)-1] { sum += v } - baseline = float64(sum) / float64(len(perMin)-1) + baseline = float64(sum) / float64(len(complete)-1) return current, baseline, true } diff --git a/backend/pkg/analytics/detector_test.go b/backend/pkg/analytics/detector_test.go index 0bed7a0..375bbaf 100644 --- a/backend/pkg/analytics/detector_test.go +++ b/backend/pkg/analytics/detector_test.go @@ -15,14 +15,17 @@ func mkMetrics(perMin []int64, p99 float64, activeConns int64) Metrics { } } -// window builds a 15-slot MessagesPerMinute window: 14 baseline slots at `base`, -// current minute at `cur`. +// window builds a 15-slot MessagesPerMinute window: 13 baseline slots at `base`, +// the most recent COMPLETE minute at `cur` in slot 13, and slot 14 left at zero. +// Slot 14 is the minute currently in progress, which the detector ignores — see +// rateWindow. Placing `cur` there instead would encode the bug this helper's +// previous version was written against. func window(base, cur int64) []int64 { w := make([]int64, 15) - for i := 0; i < 14; i++ { + for i := 0; i < 13; i++ { w[i] = base } - w[14] = cur + w[13] = cur return w } @@ -195,6 +198,38 @@ func TestDetector_TrafficDropout(t *testing.T) { } } +// A busy server rotating into a new minute must not report a dropout: the freshly +// zeroed slot is the minute in progress, not silence. The runner polls every 5s +// while the window rotates every 60s, so treating the partial slot as the current +// rate fired a spurious traffic_dropout at the top of every minute — roughly 12 +// briefings an hour after debounce, each one an ElevenLabs charge. +func TestDetector_NoDropoutAtMinuteRotation(t *testing.T) { + now := time.Now() + d := NewDetector(DefaultThresholds()) + + // Steady traffic. Slot 14 is the minute in progress, partway to ~200. + before := make([]int64, 15) + for i := 0; i < 14; i++ { + before[i] = 200 + } + before[14] = 195 + + // The window rotates: every slot shifts left and the new slot 14 starts at 0. + after := make([]int64, 15) + for i := 0; i < 13; i++ { + after[i] = 200 + } + after[13] = 195 + after[14] = 0 + + if got := d.Check(mkMetrics(before, 5, 10), now); len(got) != 0 { + t.Fatalf("steady traffic should report nothing, got %v", got) + } + if got := d.Check(mkMetrics(after, 5, 10), now.Add(5*time.Second)); len(got) != 0 { + t.Fatalf("minute rotation on a busy server must not fire, got %v", got) + } +} + func TestDetector_Debounce(t *testing.T) { now := time.Now() th := DefaultThresholds() diff --git a/backend/pkg/briefing/runner_test.go b/backend/pkg/briefing/runner_test.go index 5a14801..ca3b9a5 100644 --- a/backend/pkg/briefing/runner_test.go +++ b/backend/pkg/briefing/runner_test.go @@ -53,12 +53,16 @@ func (c *capture) first() []byte { } // spikeMetrics is a snapshot that trips the rate-spike rule. +// spikeMetrics builds a window the detector reads as a rate spike: a 90/min +// baseline with 400 in slot 13, the most recent COMPLETE minute. Slot 14 is the +// minute in progress and is deliberately left at zero — the detector ignores it, +// so putting the spike there would produce a fixture that never fires. func spikeMetrics() analytics.Metrics { perMin := make([]int64, 15) - for i := 0; i < 14; i++ { + for i := 0; i < 13; i++ { perMin[i] = 90 } - perMin[14] = 400 + perMin[13] = 400 return analytics.Metrics{MessagesPerMinute: perMin} } diff --git a/backend/pkg/client/client.go b/backend/pkg/client/client.go index aa11e36..54031bc 100644 --- a/backend/pkg/client/client.go +++ b/backend/pkg/client/client.go @@ -183,18 +183,14 @@ func (c *Client) readPump() { continue } - // Set client metadata (server overrides any client-supplied values) + // Server owns every identity and ordering field. MessageID is the DynamoDB sort + // key and Timestamp is the GSI sort key, so a client-supplied value would let one + // connection overwrite another's row or pin itself to the head of room history. msg.UserID = c.userID msg.Username = c.username msg.RoomID = c.roomID - - // Enrich with server-side fields the client doesn't set - if msg.MessageID == "" { - msg.MessageID = uuid.New().String() - } - if msg.Timestamp.IsZero() { - msg.Timestamp = time.Now().UTC() - } + msg.MessageID = uuid.New().String() + msg.Timestamp = time.Now().UTC() // Validate if err := msg.Validate(); err != nil { diff --git a/backend/pkg/client/client_test.go b/backend/pkg/client/client_test.go index d0be69b..bd6bbaf 100644 --- a/backend/pkg/client/client_test.go +++ b/backend/pkg/client/client_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/epw80/chat-analytics-platform/pkg/message" + "github.com/google/uuid" "github.com/gorilla/websocket" ) @@ -429,13 +430,19 @@ func TestClient_MetadataOverride(t *testing.T) { } defer ws.Close() - // Send message with different userID/username (should be overridden) + // Send a message that forges every server-owned field. MessageID is the DynamoDB + // sort key and Timestamp is the GSI sort key: honouring either would let this + // connection overwrite another client's row or pin itself to the head of history. + forgedTimestamp := time.Date(2099, 1, 1, 0, 0, 0, 0, time.UTC) msg := &message.Message{ - Type: message.TypeChat, - UserID: "fakeUser", - Username: "FakeName", - Content: "Hello", - } + Type: message.TypeChat, + UserID: "fakeUser", + Username: "FakeName", + Content: "Hello", + MessageID: "attacker-controlled", + Timestamp: forgedTimestamp, + } + sentAt := time.Now().UTC() data, _ := msg.ToJSON() err = ws.WriteMessage(websocket.TextMessage, data) if err != nil { @@ -462,6 +469,23 @@ func TestClient_MetadataOverride(t *testing.T) { if broadcastMsg.Username != "ActualName" { t.Errorf("expected username 'ActualName', got '%s'", broadcastMsg.Username) } + + // The forged sort keys must not survive: a replayed MessageID would overwrite the + // peer's row on PutItem, and a far-future Timestamp would pin this message to the + // head of the chronological GSI permanently. + if broadcastMsg.MessageID == "attacker-controlled" { + t.Error("client-supplied messageID was honoured; server must generate it") + } + if _, err := uuid.Parse(broadcastMsg.MessageID); err != nil { + t.Errorf("expected a server-generated UUID, got '%s': %v", broadcastMsg.MessageID, err) + } + if broadcastMsg.Timestamp.Equal(forgedTimestamp) { + t.Error("client-supplied timestamp was honoured; server must generate it") + } + if broadcastMsg.Timestamp.Before(sentAt) || broadcastMsg.Timestamp.After(time.Now().UTC()) { + t.Errorf("expected a server-side timestamp within the test window, got %s", + broadcastMsg.Timestamp) + } } // mockPersister implements the Persister interface for testing. Enqueue records diff --git a/backend/pkg/hub/hub.go b/backend/pkg/hub/hub.go index 8110253..d83ba2a 100644 --- a/backend/pkg/hub/hub.go +++ b/backend/pkg/hub/hub.go @@ -153,23 +153,46 @@ func (h *Hub) Run() { } } -// Register adds a client to the hub +// Register adds a client to the hub. +// +// The send is guarded by h.done: once Run has returned nothing drains these +// channels, and an unguarded send would block the caller forever. After +// shutdown this is a no-op rather than a Close — the shutdown branch in Run has +// already closed every client it knew about, and Client.Close closes the send +// channel, so closing a late registrant risks a double close. Its writePump +// exits on the next failed ping write. func (h *Hub) Register(client any) { - if c, ok := client.(Client); ok { - h.register <- c + c, ok := client.(Client) + if !ok { + return + } + select { + case h.register <- c: + case <-h.done: } } -// Unregister removes a client from the hub +// Unregister removes a client from the hub. Guarded by h.done for the reason +// given on Register: readPump calls this from a defer, so an unguarded send +// would leak every client goroutine at shutdown. func (h *Hub) Unregister(client any) { - if c, ok := client.(Client); ok { - h.unregister <- c + c, ok := client.(Client) + if !ok { + return + } + select { + case h.unregister <- c: + case <-h.done: } } // Broadcast sends a message to all clients in the given room. +// Guarded by h.done; see Register. func (h *Hub) Broadcast(roomID string, message []byte) { - h.broadcast <- broadcastRequest{roomID: roomID, data: message} + select { + case h.broadcast <- broadcastRequest{roomID: roomID, data: message}: + case <-h.done: + } } // BroadcastAll sends a message to every connected client across all rooms. diff --git a/backend/pkg/hub/hub_test.go b/backend/pkg/hub/hub_test.go index 991ee0c..e5f7b08 100644 --- a/backend/pkg/hub/hub_test.go +++ b/backend/pkg/hub/hub_test.go @@ -286,6 +286,37 @@ func TestHub_Shutdown(t *testing.T) { } } +// After Run returns, nothing drains the register/unregister/broadcast channels. +// Unguarded sends block forever, and because readPump calls Unregister from a +// defer that leaks every client goroutine on shutdown. Every send method must +// return promptly once the hub is down. +func TestHub_SendsDoNotBlockAfterShutdown(t *testing.T) { + hub := newTestHub() + go hub.Run() + + client := newMockClient("late") + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + hub.Shutdown() + time.Sleep(10 * time.Millisecond) + + done := make(chan struct{}) + go func() { + defer close(done) + hub.Register(newMockClient("after-shutdown")) + hub.Unregister(client) + hub.Broadcast("global", []byte("post-shutdown")) + hub.BroadcastAll([]byte("post-shutdown")) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("hub send methods blocked after shutdown; client goroutines would leak") + } +} + func TestHub_UnregisterNonExistentClient(t *testing.T) { hub := newTestHub() go hub.Run() diff --git a/backend/pkg/storage/dynamodb.go b/backend/pkg/storage/dynamodb.go index dafc146..164049f 100644 --- a/backend/pkg/storage/dynamodb.go +++ b/backend/pkg/storage/dynamodb.go @@ -2,6 +2,7 @@ package storage import ( "context" + "errors" "fmt" "log/slog" "time" @@ -276,7 +277,17 @@ func (r *DynamoDBRepository) GetMessagesByUser(ctx context.Context, userID strin return messages, nil } -// EnsureTable creates the DynamoDB table if it does not already exist +// isTableNotFound reports whether err means the table is genuinely absent, as +// opposed to DynamoDB being unreachable or throttling. Treating every error as +// "absent" sends a transient blip down the create path, where CreateTable answers +// ResourceInUseException and the whole init fails. +func isTableNotFound(err error) bool { + var notFound *types.ResourceNotFoundException + return errors.As(err, ¬Found) +} + +// EnsureTable creates the DynamoDB table if it does not already exist, and does +// not return until the table is usable. func (r *DynamoDBRepository) EnsureTable(ctx context.Context) error { schema := GetTableSchema() @@ -286,9 +297,21 @@ func (r *DynamoDBRepository) EnsureTable(ctx context.Context) error { _, err := r.client.DescribeTable(descCtx, &dynamodb.DescribeTableInput{ TableName: aws.String(schema.TableName), }) - if err == nil { + switch { + case err == nil: + // The table exists, but may still be CREATING from a previous start, so + // fall through to the wait rather than returning straight away. r.logger.Info("DynamoDB table already exists", slog.String("table", schema.TableName)) - return nil + return r.waitForActive(ctx, schema.TableName) + + case isTableNotFound(err): + // Genuinely absent — create it below. + + default: + // Throttled, unreachable, credentials rejected: anything but absent. Say so + // instead of falling through to CreateTable and failing with a misleading + // ResourceInUseException. + return fmt.Errorf("failed to describe table %s: %w", schema.TableName, err) } r.logger.Info("creating DynamoDB table", slog.String("table", schema.TableName)) @@ -336,6 +359,22 @@ func (r *DynamoDBRepository) EnsureTable(ctx context.Context) error { } r.logger.Info("DynamoDB table created successfully", slog.String("table", schema.TableName)) + return r.waitForActive(ctx, schema.TableName) +} + +// waitForActive blocks until the table is ACTIVE. CreateTable returns as soon as +// the request is accepted, so against real AWS (unlike DynamoDB Local, which is +// effectively instant) the first writes would otherwise land on a CREATING table. +// The 60s ceiling fits inside the 90s init budget the server allocates. +func (r *DynamoDBRepository) waitForActive(ctx context.Context, tableName string) error { + waiter := dynamodb.NewTableExistsWaiter(r.client) + if err := waiter.Wait(ctx, &dynamodb.DescribeTableInput{ + TableName: aws.String(tableName), + }, 60*time.Second); err != nil { + return fmt.Errorf("table %s did not become active: %w", tableName, err) + } + + r.logger.Info("DynamoDB table is active", slog.String("table", tableName)) return nil } diff --git a/backend/pkg/storage/dynamodb_test.go b/backend/pkg/storage/dynamodb_test.go index 1eb2ced..5ec9dfe 100644 --- a/backend/pkg/storage/dynamodb_test.go +++ b/backend/pkg/storage/dynamodb_test.go @@ -1,7 +1,11 @@ package storage import ( + "errors" + "fmt" "testing" + + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" ) // Unit tests for storage package @@ -10,6 +14,56 @@ import ( // Integration tests (integration_test.go) will provide comprehensive coverage // with actual DynamoDB Local. +// The distinction isTableNotFound draws is what keeps a throttle or a transport +// blip from being read as "the table is absent" and sent down the create path, +// where CreateTable answers ResourceInUseException and init fails outright — +// dropping the server into "running without persistence" while DynamoDB is fine. +// +// This is unit-tested directly rather than through EnsureTable because the +// package has no AWS client mock (see the note above); the waiter and the call +// path around it are covered by the DynamoDB Local run instead. +func TestIsTableNotFound(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + { + name: "bare not-found", + err: &types.ResourceNotFoundException{}, + want: true, + }, + { + name: "wrapped not-found", + err: fmt.Errorf("describe table: %w", &types.ResourceNotFoundException{}), + want: true, + }, + { + name: "throughput exceeded is not absence", + err: &types.ProvisionedThroughputExceededException{}, + want: false, + }, + { + name: "unrelated error is not absence", + err: errors.New("dial tcp: connection refused"), + want: false, + }, + { + name: "nil is not absence", + err: nil, + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := isTableNotFound(tc.err); got != tc.want { + t.Errorf("isTableNotFound(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + func TestGetTableSchema(t *testing.T) { schema := GetTableSchema()