Revert block-broadcast fallback path
Removes the entire fallback submitter mechanism: ckpool patch 0004,
the blocksubmit package, the wiring in main.go, the config fields,
the aggregator's fallback counters and snapshot fields, the healthz
fallback fields, and the TS type fields plus the HealthBanners
fallback alert.
Reasoning: ckpool's primary bitcoind submission must remain the
single source of truth, and getting the parallel "race a fallback
during the submit" semantics right is more architectural complexity
than the marginal reliability gain justifies. The original upstream
behavior — submit to bitcoind, retry indefinitely if unavailable —
is what we want.
Kept intact:
* Submit-attempt vs confirmed counters (block_submit_attempts /
block_submits_confirmed). Useful on their own as a "did bitcoind
confirm the submission?" signal.
* HealthBanners shows submit_gap and zmq_stale only.
* /healthz exposes submit_gap, zmq_stale, etc.
* All P0 reliability work (tailer cursor, reconcile loop, reorg
detection, multi-solve guard) and other P1 (RPC retry, WS
back-pressure, ZMQ tracking, startup readiness gate).
This commit is contained in:
satoshi
@@ -1,275 +0,0 @@
|
||||
// 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
|
||||
// <logdir>/pending-blocks/<height>-<rhash>.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 {
|
||||
// Match the config default — short enough that a failed
|
||||
// submit gets recovered while the work is still relevant,
|
||||
// long enough that we don't spuriously fallback while the
|
||||
// primary's RPC is mid-handshake.
|
||||
s.Grace = 3 * 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 `<height>-<hashprefix16>.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
|
||||
}
|
||||
@@ -39,27 +39,6 @@ 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 3s — short enough that a
|
||||
// failed submit is recovered while the work is still relevant,
|
||||
// long enough that ckpool's bounded internal wait (~3s for a live
|
||||
// primary server) and a single slow round-trip don't trigger a
|
||||
// spurious fallback. Patch 0004 makes ckpool's submit return false
|
||||
// rather than spinning indefinitely, so we no longer need to wait
|
||||
// for that case to time out.
|
||||
PendingBlocksGrace time.Duration
|
||||
}
|
||||
|
||||
func FromEnv() (*Config, error) {
|
||||
@@ -75,10 +54,6 @@ 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", 3*time.Second),
|
||||
}
|
||||
|
||||
if cfg.BitcoinRPCURL == "" {
|
||||
|
||||
@@ -112,20 +112,17 @@ func (s *Server) health(w http.ResponseWriter, r *http.Request) {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
writeJSON(w, status, map[string]any{
|
||||
"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,
|
||||
"ok": overallOK,
|
||||
"ckpool": snap.CKPoolOK,
|
||||
"bitcoin": snap.BitcoinOK,
|
||||
"submit_attempts": snap.BlockSubmitAttempts,
|
||||
"submits_confirmed": snap.BlockSubmitsConfirmed,
|
||||
"submit_gap": submitGap,
|
||||
"zmq_enabled": snap.ZMQEnabled,
|
||||
"zmq_event_age_seconds": snap.LastZMQEventAge,
|
||||
"zmq_has_event": snap.HasLastZMQEvent,
|
||||
"zmq_stale": zmqStale,
|
||||
"last_error": snap.LastError,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -80,14 +80,6 @@ type Snapshot struct {
|
||||
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
|
||||
@@ -114,7 +106,6 @@ 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.
|
||||
@@ -197,13 +188,6 @@ type Aggregator struct {
|
||||
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
|
||||
@@ -265,26 +249,6 @@ func (a *Aggregator) Run(ctx context.Context, tipEvents <-chan zmqmon.TipEvent)
|
||||
}
|
||||
}
|
||||
|
||||
// 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()
|
||||
@@ -486,11 +450,6 @@ 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()
|
||||
@@ -540,13 +499,6 @@ 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 {
|
||||
|
||||
Reference in New Issue
Block a user