From a4a894e1969652cd660ffeb9b5382d89837ce5da Mon Sep 17 00:00:00 2001 From: satoshi Date: Mon, 27 Apr 2026 21:25:56 +0300 Subject: [PATCH] P1 reliability + block-broadcast fallback path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 audits / fixes: * Bitcoin Core RPC now retries up to 3 times with linear backoff on transport errors, 5xx responses, and warm-up/loading RPC errors (code -28). Hard "no" answers (block-not-found etc.) bubble up immediately so we don't mask real errors. * WebSocket hub disconnects clients that miss 6 consecutive broadcasts (~30s with the default poll cadence). Stuck readers no longer hold stale snapshots indefinitely or freeze hub state. * ZMQ subscriber freshness: aggregator records the last-event timestamp, surfaces zmq_enabled / has_last_zmq_event / last_zmq_event_age in the snapshot. /healthz flags zmq_stale when the gap exceeds 30 minutes. * /healthz expanded with submit_attempts / submits_confirmed / submit_gap, fallback_submits_total + last_fallback_*, and the zmq staleness check. Now usable as a real-world ops dashboard signal. Block-broadcast fallback (new feature): * ckpool patch 0004: hooks local_block_submit to write the raw block hex to /pending-blocks/-.hex right before invoking generator_submitblock. Unlinks on success. ckpool's normal flow is otherwise untouched. * api/internal/blocksubmit: watcher polls the dir every 5s. Files sitting longer than the grace window (default 30s, configurable) are re-broadcast through operator-supplied backup RPC URLs in sequence. Treats both null and any "duplicate*" reject reason as success (the block landed). Pre-checks the primary chain first so a stale file from a successful-but-unlinked submit gets cleaned up without bothering fallbacks. * Aggregator records each successful fallback submission as a persistent counter and surfaces it in the snapshot so the UI can show a "primary bitcoind isn't accepting submits" alert. * Config: BACKUP_RPC_URLS (comma- or newline-separated, with optional inline credentials) plus PENDING_BLOCKS_DIR and PENDING_BLOCKS_GRACE. URLs are parsed via net/url so https://user:pass@host:port/ works cleanly. The fallback is opt-in and disabled by default. Once enabled with at least one URL, a primary bitcoind outage at the moment of solving no longer means a lost block — kamado-api re-broadcasts via whichever backup the operator trusts (a second self-hosted node, an authenticated public RPC service, etc.). --- api/cmd/kamado-api/main.go | 61 ++++ api/internal/bitcoind/rpc.go | 91 +++++- api/internal/blocksubmit/submit.go | 271 ++++++++++++++++++ api/internal/config/config.go | 20 ++ api/internal/httpapi/server.go | 36 ++- api/internal/httpapi/ws.go | 31 +- api/internal/state/aggregator.go | 74 ++++- ...0004-dump-pending-block-for-fallback.patch | 80 ++++++ ckpool/patches/README.md | 32 ++- ui/src/types.ts | 9 + 10 files changed, 687 insertions(+), 18 deletions(-) create mode 100644 api/internal/blocksubmit/submit.go create mode 100644 ckpool/patches/0004-dump-pending-block-for-fallback.patch diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go index 3804e9e..1b85f33 100644 --- a/api/cmd/kamado-api/main.go +++ b/api/cmd/kamado-api/main.go @@ -11,6 +11,7 @@ import ( "fmt" "log/slog" "net/http" + "net/url" "os" "os/signal" "strconv" @@ -19,6 +20,7 @@ import ( "time" "github.com/kamadopool/kamado-api/internal/bitcoind" + "github.com/kamadopool/kamado-api/internal/blocksubmit" "github.com/kamadopool/kamado-api/internal/ckpool" "github.com/kamadopool/kamado-api/internal/config" "github.com/kamadopool/kamado-api/internal/httpapi" @@ -116,6 +118,27 @@ func main() { go agg.IngestBlockEvents(ctx, tailer.Events) go agg.IngestAttemptEvents(ctx, tailer.Attempts) + // Fallback block submitter: if ckpool's primary bitcoind doesn't + // accept a block, the patched local_block_submit leaves the raw + // hex sitting in PENDING_BLOCKS_DIR. This watcher re-broadcasts it + // via the operator-configured BACKUP_RPC_URLS list. + if cfg.PendingBlocksDir != "" { + fallbacks := parseBackupRPCs(cfg.BackupRPCURLs, log) + sub := &blocksubmit.Submitter{ + Dir: cfg.PendingBlocksDir, + Grace: cfg.PendingBlocksGrace, + Primary: rpc, + Fallbacks: fallbacks, + Log: log, + OnSuccess: func(height int64, viaURL, viaLabel string) { + agg.RecordFallbackSubmit(height, viaLabel) + }, + } + go sub.Run(ctx, 5*time.Second) + } else { + log.Info("blocksubmit fallback disabled (PENDING_BLOCKS_DIR not set)") + } + srv := &http.Server{ Addr: cfg.ListenAddr, Handler: api.Handler(), @@ -137,3 +160,41 @@ func main() { os.Exit(1) } } + +// parseBackupRPCs splits BACKUP_RPC_URLS (newline- or comma-separated) +// into FallbackTargets. Credentials are accepted inline as +// https://user:pass@host:port/. Lines beginning with '#' are comments. +// Whitespace and empty lines are ignored. +func parseBackupRPCs(raw string, log *slog.Logger) []blocksubmit.FallbackTarget { + if raw == "" { + return nil + } + separated := strings.NewReplacer(",", "\n").Replace(raw) + var out []blocksubmit.FallbackTarget + for _, line := range strings.Split(separated, "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + u, err := url.Parse(line) + if err != nil { + log.Warn("blocksubmit: invalid fallback URL, skipping", "raw", line, "err", err) + continue + } + var user, pass string + if u.User != nil { + user = u.User.Username() + pass, _ = u.User.Password() + u.User = nil + } + clean := u.String() + out = append(out, blocksubmit.FallbackTarget{ + URL: clean, + User: user, + Password: pass, + Label: u.Host, + }) + log.Info("blocksubmit fallback registered", "url", clean, "host", u.Host, "auth", user != "") + } + return out +} diff --git a/api/internal/bitcoind/rpc.go b/api/internal/bitcoind/rpc.go index 98b1966..9a5a0c4 100644 --- a/api/internal/bitcoind/rpc.go +++ b/api/internal/bitcoind/rpc.go @@ -6,9 +6,11 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" + "strings" "time" ) @@ -50,8 +52,35 @@ func (e *rpcError) Error() string { return fmt.Sprintf("bitcoind rpc error %d: %s", e.Code, e.Message) } -// Call performs a single JSON-RPC request and unmarshals the result. +// Call performs a JSON-RPC request and unmarshals the result. Transient +// failures (transport errors, 502/503/504, or RPC errors with an +// "in warmup"/"loading"/"verifying" message that bitcoind returns +// during startup) are retried up to twice with a short backoff. RPC +// errors with semantic codes (e.g. block-not-found) are returned +// immediately because retrying won't change the answer. func (c *RPC) Call(ctx context.Context, method string, params []any, out any) error { + const maxAttempts = 3 + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + err := c.callOnce(ctx, method, params, out) + if err == nil { + return nil + } + lastErr = err + if !isRetryable(err) || attempt == maxAttempts || ctx.Err() != nil { + return err + } + backoff := time.Duration(attempt) * 200 * time.Millisecond + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + } + return lastErr +} + +func (c *RPC) callOnce(ctx context.Context, method string, params []any, out any) error { body, err := json.Marshal(rpcRequest{ JSONRPC: "1.0", ID: "kamado", @@ -78,10 +107,14 @@ func (c *RPC) Call(ctx context.Context, method string, params []any, out any) er if err != nil { return fmt.Errorf("bitcoind rpc: read body: %w", err) } - // bitcoind returns 500 on rpc errors but still with a valid JSON body. + // 5xx without a parseable body: surface as transport-level error + // so isRetryable() can flag it. var rr rpcResponse - if err := json.Unmarshal(raw, &rr); err != nil { - return fmt.Errorf("bitcoind rpc: unmarshal (status %d): %w: %s", resp.StatusCode, err, string(raw)) + if jerr := json.Unmarshal(raw, &rr); jerr != nil { + if resp.StatusCode >= 500 { + return fmt.Errorf("bitcoind rpc: %d: %s", resp.StatusCode, truncate(string(raw), 256)) + } + return fmt.Errorf("bitcoind rpc: unmarshal (status %d): %w: %s", resp.StatusCode, jerr, truncate(string(raw), 256)) } if rr.Error != nil { return rr.Error @@ -92,6 +125,56 @@ func (c *RPC) Call(ctx context.Context, method string, params []any, out any) er return nil } +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} + +// isRetryable distinguishes "the RPC didn't reach a definitive answer +// yet" (transport-level errors, 5xx, bitcoind warm-up/loading) from +// "bitcoind answered, the answer is no" (RPC error with a code, e.g. +// -5 block not found). We retry the first kind and bubble the second. +func isRetryable(err error) bool { + if err == nil { + return false + } + // rpcError values come from bitcoind itself — check for warmup/ + // loading messages that mean "ask again in a moment". + var rerr *rpcError + if errors.As(err, &rerr) { + // -28 is RPC_IN_WARMUP per bitcoin/src/rpc/protocol.h + if rerr.Code == -28 { + return true + } + msg := strings.ToLower(rerr.Message) + if strings.Contains(msg, "warming up") || + strings.Contains(msg, "loading") || + strings.Contains(msg, "verifying") || + strings.Contains(msg, "rewinding") || + strings.Contains(msg, "still busy") { + return true + } + return false + } + // Wrapped transport / 5xx / read errors all flow through fmt.Errorf + // with the "bitcoind rpc:" prefix and no rpcError target. + msg := strings.ToLower(err.Error()) + switch { + case strings.Contains(msg, "connection refused"), + strings.Contains(msg, "connection reset"), + strings.Contains(msg, "no such host"), + strings.Contains(msg, "deadline exceeded"), + strings.Contains(msg, "i/o timeout"), + strings.Contains(msg, "eof"), + strings.Contains(msg, "broken pipe"), + strings.Contains(msg, " 502"), strings.Contains(msg, " 503"), strings.Contains(msg, " 504"): + return true + } + return false +} + // ---- typed method wrappers ---- type BlockchainInfo struct { diff --git a/api/internal/blocksubmit/submit.go b/api/internal/blocksubmit/submit.go new file mode 100644 index 0000000..43f91f7 --- /dev/null +++ b/api/internal/blocksubmit/submit.go @@ -0,0 +1,271 @@ +// Package blocksubmit watches a sidecar directory that ckpool fills +// with raw block hex right before it tries to submit to its primary +// bitcoind, and re-submits via operator-configured fallback RPC URLs +// if a file persists longer than the grace period. +// +// ckpool's local_block_submit (patched in 0004) writes +// /pending-blocks/-.hex before calling +// generator_submitblock, and unlinks on success. So the rule is +// simple: any file still present after `grace` has likely failed to +// land, and we should help. +// +// Failure modes this protects against: +// - bitcoind crashed / OOM-killed at the moment of submission +// - bitcoind RPC saturated, returning timeouts +// - upstream peering issue making bitcoind unable to relay +// - ckpool itself crashes after the file was written but before unlink +// +// Each fallback URL is tried in order. submitblock returns either null +// (accepted), a "duplicate*" reason (already on chain — same outcome), +// or a hard error. We treat both null and any "duplicate" prefix as +// success and unlink. +package blocksubmit + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" + "time" + + "github.com/kamadopool/kamado-api/internal/bitcoind" +) + +// FallbackTarget is one back-up RPC endpoint we'll try if the primary +// bitcoind didn't accept a block. URL must be a full http(s):// endpoint; +// User and Password are optional (some self-hosted nodes accept +// unauthenticated localhost RPC, public RPC services usually don't). +type FallbackTarget struct { + URL string + User string + Password string + Label string // human-friendly name for logs; defaults to host +} + +// Submitter polls a pending-blocks directory and re-broadcasts blocks +// that have been sitting too long to its primary bitcoind plus any +// configured fallbacks. Stateless across iterations — the filesystem +// is the queue. +type Submitter struct { + Dir string + Grace time.Duration // how long we wait before considering a file stale + MaxAge time.Duration // older than this: log error and remove (operator intervention required) + Primary *bitcoind.RPC // primary bitcoind for "is the block already on chain?" sanity check + Fallbacks []FallbackTarget + Log *slog.Logger + + // OnSuccess, if set, is called with the height and the URL string + // that accepted the block, so the aggregator can bump a counter or + // emit a UI banner. + OnSuccess func(height int64, viaURL string, viaLabel string) +} + +// Run blocks until ctx is cancelled, sweeping the dir every poll. +func (s *Submitter) Run(ctx context.Context, poll time.Duration) { + if s.Dir == "" { + s.Log.Info("blocksubmit disabled (no pending-blocks dir)") + return + } + if s.Grace == 0 { + s.Grace = 30 * time.Second + } + if s.MaxAge == 0 { + s.MaxAge = 24 * time.Hour + } + if poll == 0 { + poll = 5 * time.Second + } + + // Ensure the directory exists so the first sweep doesn't WARN. Don't + // fail if we can't create it — ckpool may not have run yet, or perms + // may need a moment to settle. The sweep itself logs at DEBUG when + // the dir is missing. + _ = os.MkdirAll(s.Dir, 0750) + + t := time.NewTicker(poll) + defer t.Stop() + s.Log.Info("blocksubmit started", "dir", s.Dir, "grace", s.Grace, "fallbacks", len(s.Fallbacks)) + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.sweep(ctx) + } + } +} + +func (s *Submitter) sweep(ctx context.Context) { + entries, err := os.ReadDir(s.Dir) + if err != nil { + if !errors.Is(err, os.ErrNotExist) { + s.Log.Debug("blocksubmit: readdir failed", "dir", s.Dir, "err", err) + } + return + } + now := time.Now() + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".hex") { + continue + } + path := filepath.Join(s.Dir, e.Name()) + info, err := e.Info() + if err != nil { + continue + } + age := now.Sub(info.ModTime()) + if age < s.Grace { + continue + } + if age > s.MaxAge { + s.Log.Error("blocksubmit: pending block too old, abandoning", + "path", path, "age", age) + _ = os.Remove(path) + continue + } + s.handleFile(ctx, path, e.Name(), age) + } +} + +func (s *Submitter) handleFile(ctx context.Context, path, name string, age time.Duration) { + height, hashPrefix := parseFilename(name) + logger := s.Log.With("path", path, "height", height, "hash_prefix", hashPrefix, "age", age) + + // First check: maybe the primary bitcoind quietly accepted it but + // ckpool failed to unlink (process crash between submit and unlink). + // If our primary already has a block at this height, we're done — + // no need to involve fallbacks. + if height > 0 && s.Primary != nil { + lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + hash, err := s.Primary.GetBlockHash(lookupCtx, height) + cancel() + if err == nil && hash != "" { + logger.Info("blocksubmit: block already on primary chain, removing stale pending file", + "canonical_hash", hash) + _ = os.Remove(path) + return + } + } + + hex, err := os.ReadFile(path) + if err != nil { + logger.Warn("blocksubmit: read failed", "err", err) + return + } + hexStr := strings.TrimSpace(string(hex)) + if hexStr == "" { + logger.Warn("blocksubmit: empty pending file, removing") + _ = os.Remove(path) + return + } + + if len(s.Fallbacks) == 0 { + // No fallbacks configured: surface that the block is still + // pending and let the operator deal with it. Don't spam — log + // once per minute via a coarse round of `age`. + if int(age.Minutes())%1 == 0 { + logger.Warn("blocksubmit: pending block has no fallback RPCs configured") + } + return + } + + for _, fb := range s.Fallbacks { + label := fb.Label + if label == "" { + label = fb.URL + } + submitCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + err := submitOne(submitCtx, fb, hexStr) + cancel() + if err == nil { + logger.Warn("blocksubmit: fallback accepted block", "via", label) + _ = os.Remove(path) + if s.OnSuccess != nil { + s.OnSuccess(height, fb.URL, label) + } + return + } + if isDuplicate(err) { + // Block already on chain at the fallback (it was relayed + // via P2P from somewhere else). Same outcome — done. + logger.Info("blocksubmit: fallback reports duplicate (block on chain), removing pending", + "via", label, "err", err) + _ = os.Remove(path) + if s.OnSuccess != nil { + s.OnSuccess(height, fb.URL, label) + } + return + } + logger.Warn("blocksubmit: fallback rejected", "via", label, "err", err) + } + + logger.Error("blocksubmit: ALL fallbacks failed; block remains pending for next sweep", + "fallbacks_tried", len(s.Fallbacks)) +} + +// submitOne sends submitblock to one fallback RPC and returns nil on +// acceptance, a wrapped error otherwise. +func submitOne(ctx context.Context, fb FallbackTarget, hexBlock string) error { + rpc := bitcoind.NewRPC(fb.URL, fb.User, fb.Password, 30*time.Second) + // submitblock returns: null on success, or a JSON string with the + // reject reason. We have to be a bit clever with json.RawMessage to + // distinguish those. + var raw json.RawMessage + if err := rpc.Call(ctx, "submitblock", []any{hexBlock}, &raw); err != nil { + return fmt.Errorf("submitblock rpc: %w", err) + } + // Trim whitespace from the raw bytes for comparison. + body := strings.TrimSpace(string(raw)) + if body == "" || body == "null" { + return nil + } + // Strip surrounding quotes from the JSON string. + if len(body) >= 2 && body[0] == '"' && body[len(body)-1] == '"' { + body = body[1 : len(body)-1] + } + return errors.New(body) +} + +// isDuplicate covers Bitcoin Core's submitblock reject reasons that +// mean "this block is already in the chain": "duplicate", +// "duplicate-invalid", "duplicate-inconclusive". From the operator's +// perspective these are all "we don't need to keep trying". +func isDuplicate(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "duplicate") +} + +// parseFilename extracts (height, hashPrefix) from the ckpool-written +// filename `-.hex`. Returns zeros if the format +// doesn't match — the caller should still attempt the submit; we just +// can't do the "already on chain" optimization. +func parseFilename(name string) (int64, string) { + base := strings.TrimSuffix(name, ".hex") + dash := strings.IndexByte(base, '-') + if dash <= 0 { + return 0, "" + } + height, err := parseInt64(base[:dash]) + if err != nil { + return 0, base[dash+1:] + } + return height, base[dash+1:] +} + +func parseInt64(s string) (int64, error) { + var n int64 + for _, c := range s { + if c < '0' || c > '9' { + return 0, fmt.Errorf("not a number: %q", s) + } + n = n*10 + int64(c-'0') + } + return n, nil +} diff --git a/api/internal/config/config.go b/api/internal/config/config.go index 769049a..18266be 100644 --- a/api/internal/config/config.go +++ b/api/internal/config/config.go @@ -39,6 +39,22 @@ type Config struct { // the UI uses the public mempool.space; non-empty means the user // has pointed Kamado at their own instance via the StartOS config. MempoolBaseURL string + + // Directory ckpool's patched local_block_submit writes raw block + // hex into. Empty disables the fallback submitter. + PendingBlocksDir string + + // Comma- or newline-separated list of fallback bitcoind RPC URLs + // that the submitter will try in order if a block stays pending + // past the grace window. Each entry can encode credentials inline: + // https://user:pass@host:8332/ + // or without (e.g. an unauthenticated localhost backup node). + BackupRPCURLs string + + // How long a pending block file must persist before the submitter + // will try fallback RPCs. Defaults to 30s — long enough for + // ckpool's own retry loop and a fast ckpool->bitcoind round-trip. + PendingBlocksGrace time.Duration } func FromEnv() (*Config, error) { @@ -54,6 +70,10 @@ func FromEnv() (*Config, error) { DBPath: getenv("DB_PATH", "/var/lib/kamado/kamado.db"), PollInterval: getenvDuration("POLL_INTERVAL", 5*time.Second), MempoolBaseURL: os.Getenv("MEMPOOL_BASE_URL"), + + PendingBlocksDir: os.Getenv("PENDING_BLOCKS_DIR"), + BackupRPCURLs: os.Getenv("BACKUP_RPC_URLS"), + PendingBlocksGrace: getenvDuration("PENDING_BLOCKS_GRACE", 30*time.Second), } if cfg.BitcoinRPCURL == "" { diff --git a/api/internal/httpapi/server.go b/api/internal/httpapi/server.go index 76437e6..999f335 100644 --- a/api/internal/httpapi/server.go +++ b/api/internal/httpapi/server.go @@ -90,14 +90,42 @@ func writeJSON(w http.ResponseWriter, status int, v any) { func (s *Server) health(w http.ResponseWriter, r *http.Request) { snap := s.Agg.Snapshot() status := http.StatusOK + + // Submit-attempt gap: if we've tried to submit more blocks than have + // been confirmed, surface the gap. A non-zero gap is a strong signal + // even if everything else looks healthy. + submitGap := snap.BlockSubmitAttempts - snap.BlockSubmitsConfirmed + if submitGap < 0 { + submitGap = 0 + } + + // ZMQ freshness: only meaningful if the operator enabled it. Stale + // = no event in 30 minutes (typical mainnet block interval is 10 + // min, but spikes happen). + zmqStale := false + if snap.ZMQEnabled && snap.HasLastZMQEvent && snap.LastZMQEventAge > 1800 { + zmqStale = true + } + + overallOK := snap.CKPoolOK && snap.BitcoinOK && submitGap == 0 && !zmqStale if !snap.CKPoolOK || !snap.BitcoinOK { status = http.StatusServiceUnavailable } writeJSON(w, status, map[string]any{ - "ok": snap.CKPoolOK && snap.BitcoinOK, - "ckpool": snap.CKPoolOK, - "bitcoin": snap.BitcoinOK, - "last_error": snap.LastError, + "ok": overallOK, + "ckpool": snap.CKPoolOK, + "bitcoin": snap.BitcoinOK, + "submit_attempts": snap.BlockSubmitAttempts, + "submits_confirmed": snap.BlockSubmitsConfirmed, + "submit_gap": submitGap, + "fallback_submits_total": snap.FallbackSubmitsTotal, + "last_fallback_submit_at": snap.LastFallbackSubmitAt, + "last_fallback_via": snap.LastFallbackVia, + "zmq_enabled": snap.ZMQEnabled, + "zmq_event_age_seconds": snap.LastZMQEventAge, + "zmq_has_event": snap.HasLastZMQEvent, + "zmq_stale": zmqStale, + "last_error": snap.LastError, }) } diff --git a/api/internal/httpapi/ws.go b/api/internal/httpapi/ws.go index bfa21fb..2f9ca8f 100644 --- a/api/internal/httpapi/ws.go +++ b/api/internal/httpapi/ws.go @@ -42,8 +42,18 @@ type wsClient struct { conn net.Conn send chan []byte // serialized JSON frames pending write writeMu sync.Mutex // guards writes to conn (writer + reader-pong) + + // dropMisses counts consecutive Broadcast calls where this client's + // send channel was full. After maxDropMisses we close the connection + // instead of letting a stuck reader hold an old snapshot forever. + dropMisses int } +// maxDropMisses bounds how many back-to-back broadcasts we let the +// hub skip for one client before forcibly disconnecting it. With the +// poll cadence at 5s this is roughly 30s of unresponsiveness. +const maxDropMisses = 6 + // writeFrameLocked writes one frame, serializing writers on the client. func (c *wsClient) writeFrameLocked(opcode byte, payload []byte) error { c.writeMu.Lock() @@ -79,22 +89,33 @@ func (h *Hub) remove(c *wsClient) { // Broadcast serializes the snapshot once and enqueues it for every // subscribed client. Slow clients are dropped rather than blocking -// the hub. +// the hub. After maxDropMisses consecutive drops for a single client, +// we forcibly close its connection so a stuck consumer doesn't hold +// resources indefinitely or freeze on a stale snapshot. func (h *Hub) Broadcast(snap state.Snapshot) { payload, err := json.Marshal(snap) if err != nil { return } - h.mu.RLock() - defer h.mu.RUnlock() + var toClose []*wsClient + h.mu.Lock() for c := range h.clients { select { case c.send <- payload: + c.dropMisses = 0 default: - // Drop — the client's reader goroutine will clean up on - // the next write failure or close frame. + c.dropMisses++ + if c.dropMisses >= maxDropMisses { + toClose = append(toClose, c) + } } } + h.mu.Unlock() + for _, c := range toClose { + // Closing the conn unblocks the writer goroutine, which + // removes the client from the hub through wsReader's defer. + _ = c.conn.Close() + } } // handleWS upgrades an HTTP request to a WebSocket connection and diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go index c6d3469..af009af 100644 --- a/api/internal/state/aggregator.go +++ b/api/internal/state/aggregator.go @@ -77,9 +77,25 @@ type Snapshot struct { // log lines) and confirmed solves ("Solved and confirmed block"). // A growing gap means bitcoind is rejecting our submissions or // dropping the RPC — surface it in the UI as an alert. - BlockSubmitAttempts int64 `json:"block_submit_attempts"` + BlockSubmitAttempts int64 `json:"block_submit_attempts"` BlockSubmitsConfirmed int64 `json:"block_submits_confirmed"` + // Number of blocks the fallback submitter had to broadcast via a + // backup RPC URL because the primary bitcoind didn't accept them + // in time. Persistent counter; non-zero is a strong "investigate + // your bitcoind" signal even when everything ends up on chain. + FallbackSubmitsTotal int64 `json:"fallback_submits_total"` + LastFallbackSubmitAt int64 `json:"last_fallback_submit_at,omitempty"` // unix seconds + LastFallbackVia string `json:"last_fallback_via,omitempty"` + + // Health diagnostics for /healthz and the UI status badge. + // LastZMQEventAge is the seconds-since the last bitcoind hashblock + // frame arrived; -1 means no event seen since startup. ZMQEnabled + // is whether the user configured an endpoint at all. + ZMQEnabled bool `json:"zmq_enabled"` + LastZMQEventAge float64 `json:"last_zmq_event_age,omitempty"` // seconds; >=0 + HasLastZMQEvent bool `json:"has_last_zmq_event"` + // Health CKPoolOK bool `json:"ckpool_ok"` BitcoinOK bool `json:"bitcoin_ok"` @@ -98,6 +114,7 @@ const ( kvSubmitAttempts = "block_submit_attempts" kvSubmitsConfirmed = "block_submits_confirmed" + kvFallbackSubmits = "fallback_submits_total" // reconcileInterval is how often we sweep recent blocks looking // for missing hash/reward enrichment and reorg-orphaned hashes. @@ -173,6 +190,18 @@ type Aggregator struct { blockSubmitAttempts int64 blockSubmitsConfirmed int64 lastSubmitCountSave time.Time + + // Fallback-submitter telemetry. fallbackSubmits is the count of + // times a backup RPC URL successfully broadcast a block our primary + // couldn't. Persisted; survives restarts. + fallbackSubmits int64 + lastFallbackSubmitAt time.Time + lastFallbackVia string + + // ZMQ diagnostics: timestamp of the last hashblock frame relayed + // from zmqmon. Used by /healthz to flag stale subscriptions. + zmqEnabled bool + lastZMQEventTime time.Time } func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { @@ -190,6 +219,9 @@ func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog // loaded from the store before the first refresh. If tipEvents is non-nil, // each received tip triggers an immediate refresh outside the poll cadence. func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent) { + a.mu.Lock() + a.zmqEnabled = tipEvents != nil + a.mu.Unlock() a.loadPersistedBlocks() a.loadPersistedState() a.refresh(ctx) @@ -206,12 +238,35 @@ func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent) tipEvents = nil continue } + a.mu.Lock() + a.lastZMQEventTime = ev.SeenAt + a.mu.Unlock() a.Log.Debug("zmq tip, refreshing", "hash", ev.Hash) a.refresh(ctx) } } } +// RecordFallbackSubmit is called by blocksubmit.Submitter when a backup +// RPC URL accepted a block our primary bitcoind didn't. Bumps the +// persistent counter and stores the via-label so the UI can render a +// "fallback used" alert. +func (a *Aggregator) RecordFallbackSubmit(height int64, via string) { + a.mu.Lock() + a.fallbackSubmits++ + a.lastFallbackSubmitAt = time.Now() + a.lastFallbackVia = via + n := a.fallbackSubmits + a.mu.Unlock() + a.Log.Warn("fallback submission succeeded — investigate primary bitcoind", + "height", height, "via", via, "fallback_total", n) + if a.Store != nil { + if err := a.Store.SetKV(kvFallbackSubmits, strconv.FormatInt(n, 10)); err != nil { + a.Log.Warn("fallback counter persist failed", "err", err) + } + } +} + // Snapshot returns a copy of the current snapshot. func (a *Aggregator) Snapshot() Snapshot { a.mu.RLock() @@ -413,6 +468,16 @@ func (a *Aggregator) refresh(ctx context.Context) { } next.BlockSubmitAttempts = a.blockSubmitAttempts next.BlockSubmitsConfirmed = a.blockSubmitsConfirmed + next.FallbackSubmitsTotal = a.fallbackSubmits + if !a.lastFallbackSubmitAt.IsZero() { + next.LastFallbackSubmitAt = a.lastFallbackSubmitAt.Unix() + next.LastFallbackVia = a.lastFallbackVia + } + next.ZMQEnabled = a.zmqEnabled + if !a.lastZMQEventTime.IsZero() { + next.LastZMQEventAge = time.Since(a.lastZMQEventTime).Seconds() + next.HasLastZMQEvent = true + } a.snap = next cb := a.OnRefresh pushed := next @@ -456,6 +521,13 @@ func (a *Aggregator) loadPersistedState() { a.mu.Unlock() } } + if v, err := a.Store.GetKV(kvFallbackSubmits); err == nil && v != "" { + if n, perr := strconv.ParseInt(v, 10, 64); perr == nil { + a.mu.Lock() + a.fallbackSubmits = n + a.mu.Unlock() + } + } cutoff := time.Now().Add(-24 * time.Hour).Unix() if samples, err := a.Store.HashrateSince(cutoff); err != nil { diff --git a/ckpool/patches/0004-dump-pending-block-for-fallback.patch b/ckpool/patches/0004-dump-pending-block-for-fallback.patch new file mode 100644 index 0000000..c8ccb16 --- /dev/null +++ b/ckpool/patches/0004-dump-pending-block-for-fallback.patch @@ -0,0 +1,80 @@ +diff --git a/src/stratifier.c b/src/stratifier.c +index 52da790..815eb01 100644 +--- a/src/stratifier.c ++++ b/src/stratifier.c +@@ -2069,16 +2069,64 @@ process_block(const workbase_t *wb, const char *coinbase, const int cblen, + return gbt_block; + } + ++/* Write the raw block hex to a sidecar file under /pending-blocks/ ++ * before submitting, so an external watcher (kamado-api) can re-broadcast ++ * via fallback RPC nodes if our primary bitcoind doesn't accept it. The ++ * file is unlinked once generator_submitblock returns success. Best- ++ * effort: any error here is logged at INFO and never blocks the submit. */ ++static void kamado_dump_pending_block(ckpool_t *ckp, const char *gbt_block, ++ const char *rhash, int height, ++ char *out_path, size_t out_path_len) ++{ ++ char dir[512] = {}; ++ int fd; ++ size_t blen; ++ ssize_t w; ++ ++ out_path[0] = '\0'; ++ if (!ckp || !ckp->logdir || !gbt_block) ++ return; ++ snprintf(dir, sizeof(dir), "%spending-blocks", ckp->logdir); ++ if (mkdir(dir, 0750) < 0 && errno != EEXIST) { ++ LOGINFO("kamado: mkdir %s failed: %s", dir, strerror(errno)); ++ return; ++ } ++ snprintf(out_path, out_path_len, "%s/%d-%.16s.hex", dir, height, rhash); ++ fd = open(out_path, O_WRONLY | O_CREAT | O_TRUNC, 0640); ++ if (fd < 0) { ++ LOGINFO("kamado: open %s failed: %s", out_path, strerror(errno)); ++ out_path[0] = '\0'; ++ return; ++ } ++ blen = strlen(gbt_block); ++ w = write(fd, gbt_block, blen); ++ close(fd); ++ if (w != (ssize_t)blen) { ++ LOGINFO("kamado: short write to %s (%zd/%zu)", out_path, w, blen); ++ unlink(out_path); ++ out_path[0] = '\0'; ++ return; ++ } ++ LOGNOTICE("kamado: dumped pending block height %d to %s (%zu bytes)", ++ height, out_path, blen); ++} ++ + /* Submit block data locally, absorbing and freeing gbt_block */ + static bool local_block_submit(ckpool_t *ckp, char *gbt_block, const uchar *flip32, int height) + { +- bool ret = generator_submitblock(ckp, gbt_block); + char heighthash[68] = {}, rhash[68] = {}; ++ char pending_path[512] = {}; + uchar swap256[32]; ++ bool ret; + +- free(gbt_block); + swap_256(swap256, flip32); + __bin2hex(rhash, swap256, 32); ++ kamado_dump_pending_block(ckp, gbt_block, rhash, height, ++ pending_path, sizeof(pending_path)); ++ ++ ret = generator_submitblock(ckp, gbt_block); ++ ++ free(gbt_block); + generator_preciousblock(ckp, rhash); + + /* Check failures that may be inconclusive but were submitted via other +@@ -2099,6 +2147,8 @@ static bool local_block_submit(ckpool_t *ckp, char *gbt_block, const uchar *flip + height, ret ? "ACCEPTED" : "REJECTED"); + } + } ++ if (ret && pending_path[0]) ++ unlink(pending_path); + return ret; + } + diff --git a/ckpool/patches/README.md b/ckpool/patches/README.md index 15ac75f..231af00 100644 --- a/ckpool/patches/README.md +++ b/ckpool/patches/README.md @@ -9,11 +9,15 @@ Patches are applied in alphabetical order by filename. Use a numeric prefix to e ## Current state -One Kamado patch is applied on top of the pinned upstream commit: +Four Kamado patches are applied on top of the pinned upstream commit, in +alphabetical order: -| Patch | What it does | -| --------------------------------------------- | -------------------------------------------------------------------------------------- | -| `0001-expose-bestever-in-runtime-json.patch` | Adds `bestever` field to the `users` / `workers` runtime socket JSON | +| Patch | What it does | +| -------------------------------------------------------- | --------------------------------------------------------------------------------------------- | +| `0001-expose-bestever-in-runtime-json.patch` | Adds `bestever` field to the `users` / `workers` runtime socket JSON | +| `0002-enable-socket-api-responses.patch` | Always reply on the listener socket so kamado-api gets responses even with `btcsolo: true` | +| `0003-share-error-as-stratum-array.patch` | Maps `share_err` to Stratum spec error codes; emits `[code, msg, null]` per Slush | +| `0004-dump-pending-block-for-fallback.patch` | Writes the raw block hex to `/pending-blocks/` before submit; unlinks on success | ### Why 0001 matters @@ -30,6 +34,26 @@ This patch adds `bestever` to the runtime JSON so the UI can show both "this round" and "all-time" best share side by side. No behavioral change to share validation or block handling. Candidate for upstreaming. +### Why 0004 matters + +Block submission to bitcoind is the most revenue-critical RPC call ckpool +makes. If bitcoind is unreachable when a share meets network difficulty, +ckpool's `generator` thread retries indefinitely against the same single +endpoint — and the raw block data lives only in stratifier memory, so a +ckpool crash before bitcoind comes back permanently loses the block. + +This patch hooks `local_block_submit` to write the raw block hex to +`/pending-blocks/-.hex` *before* invoking +`generator_submitblock`, and unlinks the file on success. `kamado-api` +runs a watcher over that directory: if a file persists past a grace +period (default 30 s), it submits the block via fallback RPC URLs the +operator has configured. Multiple fallbacks are tried in sequence; the +file is unlinked when any fallback returns success or "duplicate" +(meaning the block already landed). + +This is a Kamado-specific integration hook — almost certainly not +upstreamable, but minimal-impact on existing ckpool behavior. + Beyond this patch, the pinned upstream commit (`cfb0f83b`, tagged as version 1.0) already includes every fix that Bassin issue #29 asked to backport, plus several improvements: diff --git a/ui/src/types.ts b/ui/src/types.ts index c5ea6be..bd6e8f2 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -132,6 +132,15 @@ export type Snapshot = { // submissions are being rejected by bitcoind. block_submit_attempts: number; block_submits_confirmed: number; + // Fallback submitter telemetry: how often we had to broadcast via a + // backup RPC URL because the primary bitcoind didn't accept in time. + fallback_submits_total: number; + last_fallback_submit_at?: number; + last_fallback_via?: string; + // ZMQ subscriber freshness diagnostics. + zmq_enabled: boolean; + has_last_zmq_event: boolean; + last_zmq_event_age?: number; // seconds ckpool_ok: boolean; bitcoin_ok: boolean; last_error?: string;