From 0a40b8f84f95c78a9a3cb364bbe3b41e1814d10b Mon Sep 17 00:00:00 2001 From: satoshi Date: Mon, 13 Apr 2026 02:53:27 +0300 Subject: [PATCH] Phase 2b: log tailer, block history, stdlib WebSocket push Adds real-time block detection via ckpool log tailing and a push channel for the upcoming Svelte UI, all stdlib-only: - logmon.Tailer follows ckpool.log with rotation/truncation survival (inode + size tracking) and parses "Solved and confirmed block N" into BlockEvent values. - state.Aggregator grows a 256-entry block ring, an OnRefresh hook, and IngestBlockEvents which best-effort enriches events with the block hash via bitcoind getblockhash. - httpapi.Hub implements RFC 6455 from scratch (SHA1 handshake, unmasked text frames out, masked frames in, ping keepalive, per-client write mutex, slow-client drop) so we don't pull in a ws dependency before we can go mod tidy. - New routes: GET /api/blocks and GET /api/ws. Snapshot pushes fire on every poll tick and immediately on block-solve. ZMQ hashblock subscription and SQLite persistence are deferred to Phase 2b.5 once the s9pk packaging repo exists and we have a real build environment for adding Go deps. --- README.md | 23 ++- api/cmd/kamado-api/main.go | 13 +- api/internal/bitcoind/rpc.go | 9 + api/internal/httpapi/server.go | 9 +- api/internal/httpapi/ws.go | 291 ++++++++++++++++++++++++++++++ api/internal/logmon/inode_unix.go | 17 ++ api/internal/logmon/tailer.go | 174 ++++++++++++++++++ api/internal/state/aggregator.go | 24 ++- api/internal/state/blocks.go | 87 +++++++++ 9 files changed, 633 insertions(+), 14 deletions(-) create mode 100644 api/internal/httpapi/ws.go create mode 100644 api/internal/logmon/inode_unix.go create mode 100644 api/internal/logmon/tailer.go create mode 100644 api/internal/state/blocks.go diff --git a/README.md b/README.md index 954b99e..c7df701 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,8 @@ Three main components: - [x] **Phase 1** — Fork & fix CKPool, build infrastructure - [~] **Phase 2** — Go API middleware - [x] Phase 2a: CKPool socket client, bitcoind RPC, state aggregator, REST API - - [ ] Phase 2b: ZMQ block notifier, WebSocket push, SQLite persistence, log tailer + - [x] Phase 2b: CKPool log tailer, block history, stdlib WebSocket push + - [ ] Phase 2b.5: ZMQ block notifier, SQLite persistence (deferred until s9pk repo exists — need real Go build env for new deps) - [ ] **Phase 3** — Svelte UI dashboard - [ ] **Phase 4** — Monorepo Docker build, full stack integration - [ ] **Phase 5** — Testing (regtest, testnet4), polish @@ -54,16 +55,18 @@ curl localhost:8080/api/health curl localhost:8080/api/pool ``` -REST endpoints (Phase 2a): +REST endpoints: -| Route | Returns | -| -------------------- | ---------------------------------------------------- | -| `GET /api/health` | CKPool + bitcoind health | -| `GET /api/pool` | Pool stats + derived hashrate windows + chain info | -| `GET /api/users` | All users from `users` socket command | -| `GET /api/workers` | All workers from `workers` socket command | -| `GET /api/clients` | All connected stratum sessions (useragent, IP, diff) | -| `GET /api/snapshot` | Full merged snapshot (everything) | +| Route | Returns | +| -------------------- | ----------------------------------------------------------- | +| `GET /api/health` | CKPool + bitcoind health | +| `GET /api/pool` | Pool stats + derived hashrate windows + chain info | +| `GET /api/users` | All users from `users` socket command | +| `GET /api/workers` | All workers from `workers` socket command | +| `GET /api/clients` | All connected stratum sessions (useragent, IP, diff) | +| `GET /api/blocks` | Recent solved blocks (in-memory ring, SQLite in Phase 2b.5) | +| `GET /api/snapshot` | Full merged snapshot (everything) | +| `GET /api/ws` | WebSocket push: full snapshot on every refresh + on solve | StartOS packaging lives in a separate repository. diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index b0be8cc..db7997f 100644 --- a/api/cmd/kamado-api/main.go +++ b/api/cmd/kamado-api/main.go @@ -19,6 +19,7 @@ import ( "github.com/kamadopool/kamado-api/internal/ckpool" "github.com/kamadopool/kamado-api/internal/config" "github.com/kamadopool/kamado-api/internal/httpapi" + "github.com/kamadopool/kamado-api/internal/logmon" "github.com/kamadopool/kamado-api/internal/state" ) @@ -45,9 +46,19 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() + api := httpapi.New(agg, log) + + // Wire snapshot refreshes into the WebSocket hub so subscribers get + // real-time updates without polling. + agg.OnRefresh = api.Hub.Broadcast + go agg.Run(ctx) - api := httpapi.New(agg, log) + // Tail the ckpool log for block-solve events. + tailer := logmon.New(cfg.CKPoolLogFile, log) + go tailer.Run(ctx) + go agg.IngestBlockEvents(ctx, tailer.Events) + srv := &http.Server{ Addr: cfg.ListenAddr, Handler: api.Handler(), diff --git a/api/internal/bitcoind/rpc.go b/api/internal/bitcoind/rpc.go index 220ff54..b911484 100644 --- a/api/internal/bitcoind/rpc.go +++ b/api/internal/bitcoind/rpc.go @@ -113,6 +113,15 @@ func (c *RPC) GetBlockchainInfo(ctx context.Context) (*BlockchainInfo, error) { return &out, nil } +// GetBlockHash returns the block hash at the given height. +func (c *RPC) GetBlockHash(ctx context.Context, height int64) (string, error) { + var out string + if err := c.Call(ctx, "getblockhash", []any{height}, &out); err != nil { + return "", err + } + return out, nil +} + // NetworkHashPS returns the network hashrate at the given block height. // `blocks` is a window (default 120). Pass -1 to use the default. func (c *RPC) GetNetworkHashPS(ctx context.Context, blocks, height int) (float64, error) { diff --git a/api/internal/httpapi/server.go b/api/internal/httpapi/server.go index c3c614f..9b1e50a 100644 --- a/api/internal/httpapi/server.go +++ b/api/internal/httpapi/server.go @@ -12,11 +12,12 @@ import ( type Server struct { Agg *state.Aggregator + Hub *Hub Log *slog.Logger } func New(agg *state.Aggregator, log *slog.Logger) *Server { - return &Server{Agg: agg, Log: log} + return &Server{Agg: agg, Hub: NewHub(), Log: log} } // Handler returns an http.Handler with all kamado routes mounted under /api. @@ -27,7 +28,9 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/users", s.users) mux.HandleFunc("GET /api/workers", s.workers) mux.HandleFunc("GET /api/clients", s.clients) + mux.HandleFunc("GET /api/blocks", s.blocks) mux.HandleFunc("GET /api/snapshot", s.snapshot) + mux.HandleFunc("GET /api/ws", s.handleWS) return mux } @@ -83,3 +86,7 @@ func (s *Server) workers(w http.ResponseWriter, r *http.Request) { func (s *Server) clients(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, s.Agg.Snapshot().Clients) } + +func (s *Server) blocks(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.Agg.Blocks()) +} diff --git a/api/internal/httpapi/ws.go b/api/internal/httpapi/ws.go new file mode 100644 index 0000000..bfa21fb --- /dev/null +++ b/api/internal/httpapi/ws.go @@ -0,0 +1,291 @@ +// Minimal RFC 6455 WebSocket server implementation, stdlib-only. +// +// Scope: push JSON snapshots to subscribed clients. We never expect +// client payloads larger than a ping/pong and we don't negotiate +// extensions (compression, fragmentation). Anything fancier is out of +// scope for Phase 2b — upgrade to a real ws library once we can run +// `go mod tidy` in a real build environment. +package httpapi + +import ( + "bufio" + "crypto/sha1" + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "io" + "net" + "net/http" + "strings" + "sync" + "time" + + "github.com/kamadopool/kamado-api/internal/state" +) + +// wsGUID is the fixed magic string from RFC 6455 §1.3 used to compute +// Sec-WebSocket-Accept. +const wsGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +// WebSocket frame opcodes we care about. +const ( + opText byte = 0x1 + opBinary byte = 0x2 + opClose byte = 0x8 + opPing byte = 0x9 + opPong byte = 0xA +) + +// wsClient is one connected WebSocket subscriber. +type wsClient struct { + conn net.Conn + send chan []byte // serialized JSON frames pending write + writeMu sync.Mutex // guards writes to conn (writer + reader-pong) +} + +// writeFrameLocked writes one frame, serializing writers on the client. +func (c *wsClient) writeFrameLocked(opcode byte, payload []byte) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + _ = c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + return writeFrame(c.conn, opcode, payload) +} + +// Hub fans out snapshot updates to all subscribed WebSocket clients. +// Register via Add / Remove from the websocket handler; Broadcast is +// called from the aggregator's OnRefresh hook. +type Hub struct { + mu sync.RWMutex + clients map[*wsClient]struct{} +} + +func NewHub() *Hub { return &Hub{clients: make(map[*wsClient]struct{})} } + +func (h *Hub) add(c *wsClient) { + h.mu.Lock() + h.clients[c] = struct{}{} + h.mu.Unlock() +} + +func (h *Hub) remove(c *wsClient) { + h.mu.Lock() + if _, ok := h.clients[c]; ok { + delete(h.clients, c) + close(c.send) + } + h.mu.Unlock() +} + +// Broadcast serializes the snapshot once and enqueues it for every +// subscribed client. Slow clients are dropped rather than blocking +// the hub. +func (h *Hub) Broadcast(snap state.Snapshot) { + payload, err := json.Marshal(snap) + if err != nil { + return + } + h.mu.RLock() + defer h.mu.RUnlock() + for c := range h.clients { + select { + case c.send <- payload: + default: + // Drop — the client's reader goroutine will clean up on + // the next write failure or close frame. + } + } +} + +// handleWS upgrades an HTTP request to a WebSocket connection and +// subscribes it to the hub. On error before the hijack, it writes a +// plain HTTP 4xx. +func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { + if !strings.EqualFold(r.Header.Get("Upgrade"), "websocket") || + !strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade") { + http.Error(w, "expected websocket upgrade", http.StatusBadRequest) + return + } + if r.Header.Get("Sec-WebSocket-Version") != "13" { + w.Header().Set("Sec-WebSocket-Version", "13") + http.Error(w, "unsupported websocket version", http.StatusUpgradeRequired) + return + } + key := r.Header.Get("Sec-WebSocket-Key") + if key == "" { + http.Error(w, "missing Sec-WebSocket-Key", http.StatusBadRequest) + return + } + + hj, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "hijack unsupported", http.StatusInternalServerError) + return + } + conn, brw, err := hj.Hijack() + if err != nil { + http.Error(w, "hijack failed", http.StatusInternalServerError) + return + } + + // Write handshake response directly to the bufio writer. + accept := wsAcceptKey(key) + resp := "HTTP/1.1 101 Switching Protocols\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Accept: " + accept + "\r\n\r\n" + if _, err := brw.WriteString(resp); err != nil { + _ = conn.Close() + return + } + if err := brw.Flush(); err != nil { + _ = conn.Close() + return + } + + client := &wsClient{conn: conn, send: make(chan []byte, 8)} + s.Hub.add(client) + s.Log.Info("ws client connected", "remote", conn.RemoteAddr()) + + // Immediately push the current snapshot so the UI doesn't wait for + // the next tick. + if payload, err := json.Marshal(s.Agg.Snapshot()); err == nil { + select { + case client.send <- payload: + default: + } + } + + go s.wsWriter(client) + go s.wsReader(client, brw.Reader) +} + +// wsWriter pumps queued payloads as text frames until send is closed +// or a write fails. +func (s *Server) wsWriter(c *wsClient) { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case payload, ok := <-c.send: + if !ok { + _ = c.writeFrameLocked(opClose, nil) + _ = c.conn.Close() + return + } + if err := c.writeFrameLocked(opText, payload); err != nil { + _ = c.conn.Close() + return + } + case <-ticker.C: + if err := c.writeFrameLocked(opPing, nil); err != nil { + _ = c.conn.Close() + return + } + } + } +} + +// wsReader reads client frames mainly to notice close/ping and to +// drive cleanup when the connection dies. Ignores any app-level +// content since the protocol is server-push only. +func (s *Server) wsReader(c *wsClient, br *bufio.Reader) { + defer s.Hub.remove(c) + for { + op, payload, err := readFrame(br) + if err != nil { + return + } + switch op { + case opClose: + return + case opPing: + _ = c.writeFrameLocked(opPong, payload) + } + } +} + +// wsAcceptKey computes Sec-WebSocket-Accept from the client's key per +// RFC 6455 §4.2.2. +func wsAcceptKey(key string) string { + h := sha1.New() + _, _ = io.WriteString(h, key+wsGUID) + return base64.StdEncoding.EncodeToString(h.Sum(nil)) +} + +// writeFrame writes a single unmasked server frame. Only supports +// payloads up to 2^63-1 bytes which is more than we'll ever send. +func writeFrame(conn net.Conn, opcode byte, payload []byte) error { + var hdr [10]byte + hdr[0] = 0x80 | (opcode & 0x0F) // FIN=1 + n := len(payload) + var hdrLen int + switch { + case n < 126: + hdr[1] = byte(n) + hdrLen = 2 + case n < 1<<16: + hdr[1] = 126 + binary.BigEndian.PutUint16(hdr[2:4], uint16(n)) + hdrLen = 4 + default: + hdr[1] = 127 + binary.BigEndian.PutUint64(hdr[2:10], uint64(n)) + hdrLen = 10 + } + if _, err := conn.Write(hdr[:hdrLen]); err != nil { + return err + } + if n > 0 { + if _, err := conn.Write(payload); err != nil { + return err + } + } + return nil +} + +// readFrame reads one client frame. Clients MUST mask (RFC 6455 §5.3); +// we enforce that and unmask in place. +func readFrame(br *bufio.Reader) (byte, []byte, error) { + var h [2]byte + if _, err := io.ReadFull(br, h[:]); err != nil { + return 0, nil, err + } + opcode := h[0] & 0x0F + masked := h[1]&0x80 != 0 + if !masked { + return 0, nil, errors.New("ws: client frame not masked") + } + length := int64(h[1] & 0x7F) + switch length { + case 126: + var ext [2]byte + if _, err := io.ReadFull(br, ext[:]); err != nil { + return 0, nil, err + } + length = int64(binary.BigEndian.Uint16(ext[:])) + case 127: + var ext [8]byte + if _, err := io.ReadFull(br, ext[:]); err != nil { + return 0, nil, err + } + length = int64(binary.BigEndian.Uint64(ext[:])) + } + // Sanity cap on client payloads; we never expect real data from + // clients. + if length > 1<<20 { + return 0, nil, errors.New("ws: client frame too large") + } + var mask [4]byte + if _, err := io.ReadFull(br, mask[:]); err != nil { + return 0, nil, err + } + payload := make([]byte, length) + if _, err := io.ReadFull(br, payload); err != nil { + return 0, nil, err + } + for i := range payload { + payload[i] ^= mask[i%4] + } + return opcode, payload, nil +} diff --git a/api/internal/logmon/inode_unix.go b/api/internal/logmon/inode_unix.go new file mode 100644 index 0000000..9f45721 --- /dev/null +++ b/api/internal/logmon/inode_unix.go @@ -0,0 +1,17 @@ +//go:build unix + +package logmon + +import ( + "os" + "syscall" +) + +// inodeOf returns the inode number for a FileInfo on unix systems, or 0 +// if the underlying stat is unavailable. Used to detect log rotation. +func inodeOf(fi os.FileInfo) uint64 { + if st, ok := fi.Sys().(*syscall.Stat_t); ok { + return st.Ino + } + return 0 +} diff --git a/api/internal/logmon/tailer.go b/api/internal/logmon/tailer.go new file mode 100644 index 0000000..99882f1 --- /dev/null +++ b/api/internal/logmon/tailer.go @@ -0,0 +1,174 @@ +// Package logmon tails the ckpool log file looking for notable events — +// primarily block-solve lines, which are the most reliable signal we have +// that a block was found, short of a ZMQ hashblock subscription. +// +// ckpool-solo logs a line like: +// +// Solved and confirmed block 840123 +// +// from stratifier.c via LOGWARNING when a submitted share passes network +// difficulty and bitcoind confirms acceptance. We parse these lines, +// emit BlockEvent values on Events, and let the aggregator enrich them +// with hash/reward via bitcoind RPC. +package logmon + +import ( + "bufio" + "context" + "errors" + "io" + "log/slog" + "os" + "regexp" + "strconv" + "time" +) + +// BlockEvent is emitted when the tailer sees a "Solved and confirmed block" +// line in the ckpool log. Hash and worker are populated later by the +// aggregator once it cross-references bitcoind. +type BlockEvent struct { + Height int64 `json:"height"` + SeenAt time.Time `json:"seen_at"` + RawLine string `json:"raw_line"` +} + +// Tailer follows a log file, surviving rotation/truncation, and emits +// parsed events. Create with New, then Run in a goroutine. +type Tailer struct { + Path string + Events chan BlockEvent + Log *slog.Logger + PollWait time.Duration // how long to sleep between EOF polls +} + +func New(path string, log *slog.Logger) *Tailer { + return &Tailer{ + Path: path, + Events: make(chan BlockEvent, 16), + Log: log, + PollWait: 500 * time.Millisecond, + } +} + +var solvedRE = regexp.MustCompile(`Solved and confirmed block\s+(\d+)`) + +// Run blocks until ctx is cancelled. It opens the file, seeks to the end, +// and reads new lines as they are appended. If the file is rotated +// (shrinks, or inode changes), it re-opens. +func (t *Tailer) Run(ctx context.Context) { + defer close(t.Events) + + var ( + f *os.File + reader *bufio.Reader + lastIno uint64 + lastPos int64 + ) + + open := func() error { + if f != nil { + _ = f.Close() + } + nf, err := os.Open(t.Path) + if err != nil { + return err + } + // Start at end on first open so we don't replay old events. + if _, err := nf.Seek(0, io.SeekEnd); err != nil { + _ = nf.Close() + return err + } + st, err := nf.Stat() + if err != nil { + _ = nf.Close() + return err + } + f = nf + reader = bufio.NewReader(f) + lastIno = inodeOf(st) + lastPos, _ = f.Seek(0, io.SeekCurrent) + return nil + } + + // Initial open; retry on failure until the file exists. + for { + if err := open(); err != nil { + t.Log.Warn("logmon: waiting for log file", "path", t.Path, "err", err) + if !sleep(ctx, 2*time.Second) { + return + } + continue + } + break + } + defer func() { + if f != nil { + _ = f.Close() + } + }() + + for { + if ctx.Err() != nil { + return + } + line, err := reader.ReadString('\n') + if len(line) > 0 { + t.handleLine(line) + lastPos, _ = f.Seek(0, io.SeekCurrent) + } + if err == nil { + continue + } + if !errors.Is(err, io.EOF) { + t.Log.Warn("logmon: read error, reopening", "err", err) + if !sleep(ctx, t.PollWait) { + return + } + _ = open() + continue + } + + // EOF: check for rotation (inode changed) or truncation (size < pos). + if st, statErr := os.Stat(t.Path); statErr == nil { + if inodeOf(st) != lastIno || st.Size() < lastPos { + t.Log.Info("logmon: log rotated, reopening", "path", t.Path) + if err := open(); err != nil { + t.Log.Warn("logmon: reopen failed", "err", err) + } + continue + } + } + if !sleep(ctx, t.PollWait) { + return + } + } +} + +func (t *Tailer) handleLine(line string) { + m := solvedRE.FindStringSubmatch(line) + if m == nil { + return + } + height, err := strconv.ParseInt(m[1], 10, 64) + if err != nil { + return + } + ev := BlockEvent{Height: height, SeenAt: time.Now(), RawLine: line} + select { + case t.Events <- ev: + t.Log.Info("logmon: block solved", "height", height) + default: + t.Log.Warn("logmon: events channel full, dropping", "height", height) + } +} + +// sleep returns false if ctx was cancelled during the wait. +func sleep(ctx context.Context, d time.Duration) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(d): + return true + } +} diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index 1b0e0a5..88407c9 100644 --- a/api/internal/state/aggregator.go +++ b/api/internal/state/aggregator.go @@ -36,6 +36,9 @@ type Snapshot struct { Chain *bitcoind.BlockchainInfo `json:"chain"` NetworkHashrateHs float64 `json:"network_hashrate_hs"` + // Recent found blocks (in-memory history; persisted in Phase 2b.5). + RecentBlocks []BlockRecord `json:"recent_blocks,omitempty"` + // Health CKPoolOK bool `json:"ckpool_ok"` BitcoinOK bool `json:"bitcoin_ok"` @@ -49,8 +52,13 @@ type Aggregator struct { Interval time.Duration Log *slog.Logger - mu sync.RWMutex - snap Snapshot + // OnRefresh, if set, is called (non-blocking) after each snapshot + // refresh. Used by the WebSocket hub to push updates to clients. + OnRefresh func(Snapshot) + + mu sync.RWMutex + snap Snapshot + blocks []BlockRecord } func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { @@ -132,6 +140,18 @@ func (a *Aggregator) refresh(ctx context.Context) { } a.mu.Lock() + // Attach current block history so /api/snapshot and WebSocket + // pushes carry the same view. + if len(a.blocks) > 0 { + next.RecentBlocks = make([]BlockRecord, len(a.blocks)) + copy(next.RecentBlocks, a.blocks) + } a.snap = next + cb := a.OnRefresh + pushed := next a.mu.Unlock() + + if cb != nil { + cb(pushed) + } } diff --git a/api/internal/state/blocks.go b/api/internal/state/blocks.go new file mode 100644 index 0000000..5297a29 --- /dev/null +++ b/api/internal/state/blocks.go @@ -0,0 +1,87 @@ +package state + +import ( + "context" + "time" + + "github.com/kamadopool/kamado-api/internal/logmon" +) + +// BlockRecord is a found block, merged from a logmon event with bitcoind +// data if available. Hash and Reward are populated best-effort via the +// RPC lookup scheduled right after the log line is seen. +type BlockRecord struct { + Height int64 `json:"height"` + Hash string `json:"hash,omitempty"` + RewardBT float64 `json:"reward_btc,omitempty"` + FoundAt time.Time `json:"found_at"` + Source string `json:"source"` // "logmon" for now; "zmq" later +} + +// IngestBlockEvents reads block events from the tailer and appends them +// to the snapshot's block history. Runs until ctx is cancelled. +func (a *Aggregator) IngestBlockEvents(ctx context.Context, events <-chan logmon.BlockEvent) { + for { + select { + case <-ctx.Done(): + return + case ev, ok := <-events: + if !ok { + return + } + rec := BlockRecord{ + Height: ev.Height, + FoundAt: ev.SeenAt, + Source: "logmon", + } + // Best-effort enrich with hash via bitcoind. + if a.RPC != nil { + lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + if hash, err := a.RPC.GetBlockHash(lookupCtx, ev.Height); err == nil { + rec.Hash = hash + } else { + a.Log.Warn("bitcoind getblockhash failed", "height", ev.Height, "err", err) + } + cancel() + } + pushed := a.appendBlock(rec) + a.Log.Info("block recorded", "height", rec.Height, "hash", rec.Hash) + // Push immediately so WebSocket clients see the solve + // without waiting for the next poll tick. + if a.OnRefresh != nil { + a.OnRefresh(pushed) + } + } + } +} + +// maxBlockHistory caps in-memory block history. Persistence comes in +// Phase 2b.5 via SQLite; for now recent blocks survive only this +// process's lifetime. +const maxBlockHistory = 256 + +// appendBlock records a new block and returns a copy of the current +// snapshot with the updated history attached, suitable for an immediate +// WebSocket broadcast. +func (a *Aggregator) appendBlock(rec BlockRecord) Snapshot { + a.mu.Lock() + defer a.mu.Unlock() + a.blocks = append(a.blocks, rec) + if len(a.blocks) > maxBlockHistory { + a.blocks = a.blocks[len(a.blocks)-maxBlockHistory:] + } + snap := a.snap + snap.RecentBlocks = make([]BlockRecord, len(a.blocks)) + copy(snap.RecentBlocks, a.blocks) + a.snap = snap + return snap +} + +// Blocks returns a copy of the recent block history, newest last. +func (a *Aggregator) Blocks() []BlockRecord { + a.mu.RLock() + defer a.mu.RUnlock() + out := make([]BlockRecord, len(a.blocks)) + copy(out, a.blocks) + return out +}