KamadoPool/api/internal/blocksubmit/submit.go

// 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 {
		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 `<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
}