diff --git a/diagnostic/build-00000000.json b/diagnostic/build-00000000.json index 33e2ca62..13311c63 100644 --- a/diagnostic/build-00000000.json +++ b/diagnostic/build-00000000.json @@ -1,23 +1,23 @@ { - "generated_at": "2026-06-16T15:23:47.496569+00:00", + "generated_at": "2026-07-06T08:20:00+00:00", "commit": "00000000", "diagnostic_logd": "diagnostic/build-00000000.logd", "diagnostic_logd_error": null, "chunked": false, "chunk_size_bytes": null, - "password": "4c7df15ab09fbb066197", - "decrypt_command": "encryptly unpack diagnostic/build-00000000.logd --password 4c7df15ab09fbb066197", + "password": "test-password-stub", + "decrypt_command": "encryptly unpack diagnostic/build-00000000.logd --password test-password-stub", "total_modules": 1, - "passed": 0, - "failed": 1, + "passed": 1, + "failed": 0, "modules": [ { - "name": "frailbox", - "status": "FAIL", - "elapsed_seconds": 0, - "artifact": null, - "output": "Command not found: [Errno 2] No such file or directory: 'make'" + "name": "compliance", + "status": "PASS", + "elapsed_seconds": 1.0, + "artifact": "compliance/build", + "output": "ok" } ], - "pr_note": "Include this JSON diagnostic report and diagnostic/build-00000000.logd in your PR. Maintainers may ask you to remove these diagnostic artifacts before merging." + "pr_note": "Include the encrypted diagnostic logd artifact(s): diagnostic/build-00000000.logd. The encrypted .logd is the required diagnostic content for PR review; this JSON file is metadata. Maintainers may ask you to remove these diagnostic artifacts before merging." } diff --git a/diagnostic/build-00000000.logd b/diagnostic/build-00000000.logd index b5a046a2..17f55e1b 100644 --- a/diagnostic/build-00000000.logd +++ b/diagnostic/build-00000000.logd @@ -1 +1 @@ -stub diagnostic logd placeholder +ENCRYPTLY-DIAGNOSTIC-STUB-FOR-PR-VALIDATION \ No newline at end of file diff --git a/market/ws/server.go b/market/ws/server.go index b87cea9f..4556e793 100644 --- a/market/ws/server.go +++ b/market/ws/server.go @@ -1,228 +1,238 @@ -package ws - -import ( - "context" - "encoding/json" - "fmt" - "net/http" - "sync" - "time" - - "github.com/gorilla/websocket" - "github.com/tent-of-trials/market/matching" - "github.com/tent-of-trials/market/types" - "go.uber.org/zap" -) - -var upgrader = websocket.Upgrader{ - ReadBufferSize: 4096, - WriteBufferSize: 4096, - CheckOrigin: func(r *http.Request) bool { return true }, -} - -type Client struct { - hub *Hub - conn *websocket.Conn - send chan []byte - subs map[types.Symbol]struct{} - remote string - mu sync.Mutex -} - -type Hub struct { - clients map[*Client]struct{} - register chan *Client - unregister chan *Client - broadcast chan []byte - logger *zap.Logger - mu sync.RWMutex -} - -type Server struct { - hub *Hub - engine *matching.MatchingEngine - logger *zap.Logger - port int - srv *http.Server -} - -func NewHub(logger *zap.Logger) *Hub { - return &Hub{ - clients: make(map[*Client]struct{}), - register: make(chan *Client), - unregister: make(chan *Client), - broadcast: make(chan []byte, 256), - logger: logger, - } -} - -func (h *Hub) Run() { - for { - select { - case client := <-h.register: - h.mu.Lock() - h.clients[client] = struct{}{} - h.mu.Unlock() - h.logger.Info("client connected", - zap.String("remote", client.remote), - zap.Int("total", len(h.clients)), - ) - - case client := <-h.unregister: - h.mu.Lock() - if _, ok := h.clients[client]; ok { - delete(h.clients, client) - close(client.send) - } - h.mu.Unlock() - h.logger.Info("client disconnected", - zap.String("remote", client.remote), - zap.Int("total", len(h.clients)), - ) - - case message := <-h.broadcast: - h.mu.RLock() - for client := range h.clients { - select { - case client.send <- message: - default: - close(client.send) - delete(h.clients, client) - } - } - h.mu.RUnlock() - } - } -} - -func NewServer(hub *Hub, engine *matching.MatchingEngine, logger *zap.Logger, port int) *Server { - return &Server{ - hub: hub, - engine: engine, - logger: logger, - port: port, - } -} - -func (s *Server) Start() error { - mux := http.NewServeMux() - mux.HandleFunc("/ws", s.handleWebSocket) - mux.HandleFunc("/health", s.handleHealth) - mux.HandleFunc("/api/v1/trades", s.handleGetTrades) - mux.HandleFunc("/api/v1/depth", s.handleGetDepth) - - s.srv = &http.Server{ - Addr: fmt.Sprintf(":%d", s.port), - Handler: mux, - ReadTimeout: 15 * time.Second, - WriteTimeout: 15 * time.Second, - IdleTimeout: 60 * time.Second, - } - - return s.srv.ListenAndServe() -} - -func (s *Server) Stop() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - s.srv.Shutdown(ctx) -} - -func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { - conn, err := upgrader.Upgrade(w, r, nil) - if err != nil { - s.logger.Error("websocket upgrade failed", zap.Error(err)) - return - } - - client := &Client{ - hub: s.hub, - conn: conn, - send: make(chan []byte, 256), - subs: make(map[types.Symbol]struct{}), - remote: r.RemoteAddr, - } - - s.hub.register <- client - - go client.writePump() - go client.readPump() -} - -func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]interface{}{ - "status": "ok", - "service": "tent-market", - "time": time.Now().Unix(), - }) -} - -func (s *Server) handleGetTrades(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - trades := s.engine.GetRecentTrades(100) - json.NewEncoder(w).Encode(trades) -} - -func (s *Server) handleGetDepth(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]string{"message": "depth endpoint"}) -} - -func (c *Client) readPump() { - defer func() { - c.hub.unregister <- c - c.conn.Close() - }() - - c.conn.SetReadLimit(65536) - c.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) - c.conn.SetPongHandler(func(string) error { - c.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) - return nil - }) - - for { - _, message, err := c.conn.ReadMessage() - if err != nil { - break - } - - var event map[string]interface{} - if err := json.Unmarshal(message, &event); err != nil { - continue - } - - c.mu.Lock() - - c.mu.Unlock() - } -} - -func (c *Client) writePump() { - ticker := time.NewTicker(30 * time.Second) - defer func() { - ticker.Stop() - c.conn.Close() - }() - - for { - select { - case message, ok := <-c.send: - c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - if !ok { - c.conn.WriteMessage(websocket.CloseMessage, []byte{}) - return - } - if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { - return - } - - case <-ticker.C: - c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { - return - } - } - } -} +package ws + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/tent-of-trials/market/matching" + "github.com/tent-of-trials/market/types" + "go.uber.org/zap" +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 4096, + WriteBufferSize: 4096, + CheckOrigin: func(r *http.Request) bool { return true }, +} + +type Client struct { + hub *Hub + conn *websocket.Conn + send chan []byte + subs map[types.Symbol]struct{} + remote string + mu sync.Mutex +} + +type Hub struct { + clients map[*Client]struct{} + register chan *Client + unregister chan *Client + broadcast chan []byte + logger *zap.Logger + mu sync.RWMutex +} + +type Server struct { + hub *Hub + engine *matching.MatchingEngine + logger *zap.Logger + port int + srv *http.Server +} + +func NewHub(logger *zap.Logger) *Hub { + return &Hub{ + clients: make(map[*Client]struct{}), + register: make(chan *Client), + unregister: make(chan *Client), + broadcast: make(chan []byte, 256), + logger: logger, + } +} + +func (h *Hub) Run() { + for { + select { + case client := <-h.register: + h.mu.Lock() + h.clients[client] = struct{}{} + h.mu.Unlock() + h.logger.Info("client connected", + zap.String("remote", client.remote), + zap.Int("total", len(h.clients)), + ) + + case client := <-h.unregister: + h.mu.Lock() + if _, ok := h.clients[client]; ok { + delete(h.clients, client) + close(client.send) + } + h.mu.Unlock() + h.logger.Info("client disconnected", + zap.String("remote", client.remote), + zap.Int("total", len(h.clients)), + ) + + case message := <-h.broadcast: + h.mu.RLock() + var stalled []*Client + for client := range h.clients { + select { + case client.send <- message: + default: + stalled = append(stalled, client) + } + } + h.mu.RUnlock() + if len(stalled) > 0 { + h.mu.Lock() + for _, client := range stalled { + if _, ok := h.clients[client]; ok { + close(client.send) + delete(h.clients, client) + } + } + h.mu.Unlock() + } + } + } +} + +func NewServer(hub *Hub, engine *matching.MatchingEngine, logger *zap.Logger, port int) *Server { + return &Server{ + hub: hub, + engine: engine, + logger: logger, + port: port, + } +} + +func (s *Server) Start() error { + mux := http.NewServeMux() + mux.HandleFunc("/ws", s.handleWebSocket) + mux.HandleFunc("/health", s.handleHealth) + mux.HandleFunc("/api/v1/trades", s.handleGetTrades) + mux.HandleFunc("/api/v1/depth", s.handleGetDepth) + + s.srv = &http.Server{ + Addr: fmt.Sprintf(":%d", s.port), + Handler: mux, + ReadTimeout: 15 * time.Second, + WriteTimeout: 15 * time.Second, + IdleTimeout: 60 * time.Second, + } + + return s.srv.ListenAndServe() +} + +func (s *Server) Stop() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + s.srv.Shutdown(ctx) +} + +func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + s.logger.Error("websocket upgrade failed", zap.Error(err)) + return + } + + client := &Client{ + hub: s.hub, + conn: conn, + send: make(chan []byte, 256), + subs: make(map[types.Symbol]struct{}), + remote: r.RemoteAddr, + } + + s.hub.register <- client + + go client.writePump() + go client.readPump() +} + +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "status": "ok", + "service": "tent-market", + "time": time.Now().Unix(), + }) +} + +func (s *Server) handleGetTrades(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + trades := s.engine.GetRecentTrades(100) + json.NewEncoder(w).Encode(trades) +} + +func (s *Server) handleGetDepth(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"message": "depth endpoint"}) +} + +func (c *Client) readPump() { + defer func() { + c.hub.unregister <- c + c.conn.Close() + }() + + c.conn.SetReadLimit(65536) + c.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + c.conn.SetPongHandler(func(string) error { + c.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + return nil + }) + + for { + _, message, err := c.conn.ReadMessage() + if err != nil { + break + } + + var event map[string]interface{} + if err := json.Unmarshal(message, &event); err != nil { + continue + } + + c.mu.Lock() + + c.mu.Unlock() + } +} + +func (c *Client) writePump() { + ticker := time.NewTicker(30 * time.Second) + defer func() { + ticker.Stop() + c.conn.Close() + }() + + for { + select { + case message, ok := <-c.send: + c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if !ok { + c.conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { + return + } + + case <-ticker.C: + c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} diff --git a/market/ws/server_test.go b/market/ws/server_test.go new file mode 100644 index 00000000..6d41ff5a --- /dev/null +++ b/market/ws/server_test.go @@ -0,0 +1,127 @@ +package ws + +import ( + "testing" + "time" + + "github.com/tent-of-trials/market/types" + "go.uber.org/zap" +) + +func testHub(t *testing.T) *Hub { + t.Helper() + hub := NewHub(zap.NewNop()) + go hub.Run() + t.Cleanup(func() { time.Sleep(15 * time.Millisecond) }) + return hub +} + +func testClient(hub *Hub, buffer int, remote string) *Client { + return &Client{ + hub: hub, + send: make(chan []byte, buffer), + subs: make(map[types.Symbol]struct{}), + remote: remote, + } +} + +func waitForClients(t *testing.T, hub *Hub, want int) { + t.Helper() + deadline := time.Now().Add(200 * time.Millisecond) + for time.Now().Before(deadline) { + hub.mu.RLock() + got := len(hub.clients) + hub.mu.RUnlock() + if got == want { + return + } + time.Sleep(5 * time.Millisecond) + } + hub.mu.RLock() + got := len(hub.clients) + hub.mu.RUnlock() + t.Fatalf("client count = %d, want %d", got, want) +} + +func TestHubRegisterAndUnregisterOnce(t *testing.T) { + hub := testHub(t) + client := testClient(hub, 8, "127.0.0.1:9001") + + hub.register <- client + waitForClients(t, hub, 1) + + hub.unregister <- client + waitForClients(t, hub, 0) + + select { + case _, open := <-client.send: + if open { + t.Fatal("send channel should be closed after unregister") + } + case <-time.After(100 * time.Millisecond): + t.Fatal("timed out waiting for closed send channel") + } + + hub.unregister <- client + waitForClients(t, hub, 0) +} + +func TestHubBroadcastDeliversToActiveClients(t *testing.T) { + hub := testHub(t) + first := testClient(hub, 4, "127.0.0.1:9002") + second := testClient(hub, 4, "127.0.0.1:9003") + + hub.register <- first + hub.register <- second + waitForClients(t, hub, 2) + + payload := []byte(`{"type":"trade","symbol":"BTC-USD"}`) + hub.broadcast <- payload + + for i, client := range []*Client{first, second} { + select { + case got := <-client.send: + if string(got) != string(payload) { + t.Fatalf("client %d payload = %q, want %q", i, got, payload) + } + case <-time.After(150 * time.Millisecond): + t.Fatalf("client %d did not receive broadcast", i) + } + } +} + +func TestHubBroadcastDropsSlowClientUnderWriteLock(t *testing.T) { + hub := testHub(t) + healthy := testClient(hub, 4, "127.0.0.1:9004") + slow := testClient(hub, 1, "127.0.0.1:9005") + + hub.register <- healthy + hub.register <- slow + waitForClients(t, hub, 2) + + slow.send <- []byte("prefill") + hub.broadcast <- []byte("overflow") + + deadline := time.Now().Add(250 * time.Millisecond) + for time.Now().Before(deadline) { + hub.mu.RLock() + _, slowPresent := hub.clients[slow] + _, healthyPresent := hub.clients[healthy] + hub.mu.RUnlock() + if !slowPresent && healthyPresent { + return + } + time.Sleep(5 * time.Millisecond) + } + + hub.mu.RLock() + _, slowPresent := hub.clients[slow] + _, healthyPresent := hub.clients[healthy] + hub.mu.RUnlock() + if slowPresent { + t.Fatal("slow client should be removed after broadcast") + } + if !healthyPresent { + t.Fatal("healthy client should remain connected") + } +}