feat(fastsync): /fastsync/anchors feed + concurrent (SQLite/WAL) bundle store (#115)

cmd/ion-node/main.go

@@ -27,6 +27,7 @@ import (
 	"github.com/13x-tech/ion-node/internal/cas"
 	"github.com/13x-tech/ion-node/internal/chain"
 	"github.com/13x-tech/ion-node/internal/config"
+	"github.com/13x-tech/ion-node/internal/fastsync"
 	"github.com/13x-tech/ion-node/internal/fee"
 	"github.com/13x-tech/ion-node/internal/index"
 	"github.com/13x-tech/ion-node/internal/lock"
@@ -296,15 +297,16 @@ func buildCAS(cfg *config.Config, logger *slog.Logger) (sidetree.CAS, error) {
 
 // node bundles the wired-up components.
 type node struct {
-	cfg    *config.Config
-	logger *slog.Logger
-	hc     *chain.HeaderChain
-	client *p2p.Client
-	store  store.Store
-	idx    *index.Indexer
-	txIdx  *txindex.Index    // resolver cache for writer/lock verification; nil when disabled
-	cas    sidetree.CAS      // cached external IPFS CAS; nil when --observe=false
-	obs    *process.Observer // nil when --observe=false
+	cfg     *config.Config
+	logger  *slog.Logger
+	hc      *chain.HeaderChain
+	client  *p2p.Client
+	store   store.Store
+	idx     *index.Indexer
+	txIdx   *txindex.Index    // resolver cache for writer/lock verification; nil when disabled
+	bundles *fastsync.Store   // fast-sync bundle capture; nil unless --serve-fast-sync
+	cas     sidetree.CAS      // cached external IPFS CAS; nil when --observe=false
+	obs     *process.Observer // nil when --observe=false
 }
 
 // writerVerification bundles the bitcoind-free resolver wiring shared by the indexer
@@ -383,8 +385,26 @@ func buildNode(cCtx *cli.Context) (*node, error) {
 		logger.Info("writer verification enabled", "esplora", endpoint)
 	}
 
+	// Fast-sync bundle capture (#115): when serving fast-sync, record a verifiable
+	// inclusion bundle for each anchor as it is indexed. SQLite/WAL so `serve` can
+	// read the bundles concurrently with this writer.
+	var bundleStore *fastsync.Store
+	if cfg.ServeFastSync {
+		bs, err := fastsync.Open(filepath.Join(cfg.DataDir, "bundles.db"))
+		if err != nil {
+			_ = st.Close()
+			if txIdx != nil {
+				_ = txIdx.Close()
+			}
+			return nil, fmt.Errorf("bundle store: %w", err)
+		}
+		bundleStore = bs
+		idxOpts = append(idxOpts, index.WithBundleWriter(bs))
+		logger.Info("fast-sync bundle capture enabled", "db", "bundles.db")
+	}
+
 	idx := index.New(cfg, client, st, logger, idxOpts...)
-	n := &node{cfg: cfg, logger: logger, hc: hc, client: client, store: st, idx: idx, txIdx: txIdx}
+	n := &node{cfg: cfg, logger: logger, hc: hc, client: client, store: st, idx: idx, txIdx: txIdx, bundles: bundleStore}
 
 	if cCtx.Bool("observe") {
 		casClient, err := buildCAS(cfg, logger)
@@ -428,6 +448,9 @@ func (n *node) close() {
 	if n.txIdx != nil {
 		_ = n.txIdx.Close()
 	}
+	if n.bundles != nil {
+		_ = n.bundles.Close()
+	}
 	_ = n.store.Close()
 }
 
@@ -735,7 +758,13 @@ func runServe(cCtx *cli.Context) error {
 	handler := api.NewHandler(resolver, logger).WithStats(st) // GET /stats: dids, pendingRetryable, heights
 	if cfg.ServeFastSync {
 		handler = handler.WithCASGateway(casClient) // GET /fastsync/cas/<cid>
-		logger.Info("fast-sync content gateway enabled", "endpoint", "/fastsync/cas/<cid>")
+		bs, err := fastsync.Open(filepath.Join(cfg.DataDir, "bundles.db"))
+		if err != nil {
+			return fmt.Errorf("bundle store: %w", err)
+		}
+		defer bs.Close()
+		handler = handler.WithAnchorBundles(fastsync.NewService(bs)) // GET /fastsync/anchors
+		logger.Info("fast-sync gateway enabled", "endpoints", "/fastsync/cas/<cid>, /fastsync/anchors?from=&to=")
 	}
 	srv := api.NewServer(cCtx.String("http-addr"), handler.Mux())
 

internal/api/resolution.go

@@ -7,15 +7,18 @@
 package api
 
 import (
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"log/slog"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/13x-tech/ion-node/internal/cas"
+	"github.com/13x-tech/ion-node/internal/fastsync"
 	"github.com/13x-tech/ion-node/internal/resolve"
 )
 
@@ -75,6 +78,16 @@ type CASGateway interface {
 	Get(cid string, maxSizeInBytes int) ([]byte, error)
 }
 
+// AnchorBundleProvider returns verifiable anchor bundles for a height range
+// (*fastsync.Service satisfies it). It backs GET /fastsync/anchors: the client
+// verifies each bundle's merkle proof against its OWN PoW headers, so this source is
+// untrusted — it can only omit, never forge.
+type AnchorBundleProvider interface {
+	AnchorBundles(from, to int32) ([]fastsync.Bundle, error)
+}
+
+const fastSyncAnchorsPath = "/fastsync/anchors"
+
 // maxGatewayBlobBytes bounds a single content-gateway response. Sidetree's largest
 // file types are well under this; it guards against an unexpectedly huge blob.
 const maxGatewayBlobBytes = 1 << 22 // 4 MiB
@@ -84,8 +97,9 @@ const fastSyncCASPrefix = "/fastsync/cas/"
 // Handler serves the DID resolution API.
 type Handler struct {
 	resolver DIDResolver
-	stats    StatsProvider // optional; when set, enables GET /stats
-	cas      CASGateway    // optional; when set, enables GET /fastsync/cas/<cid>
+	stats    StatsProvider        // optional; when set, enables GET /stats
+	cas      CASGateway           // optional; when set, enables GET /fastsync/cas/<cid>
+	anchors  AnchorBundleProvider // optional; when set, enables GET /fastsync/anchors
 	log      *slog.Logger
 }
 
@@ -113,6 +127,14 @@ func (h *Handler) WithCASGateway(g CASGateway) *Handler {
 	return h
 }
 
+// WithAnchorBundles enables GET /fastsync/anchors?from=&to=, the verifiable
+// anchor-bundle feed for fast-sync bootstrap / browser light-clients. CORS-open; the
+// bundles are self-proving against the client's own Bitcoin headers.
+func (h *Handler) WithAnchorBundles(p AnchorBundleProvider) *Handler {
+	h.anchors = p
+	return h
+}
+
 // Mux returns the routed HTTP handler: the resolution endpoint, /health, and (when
 // configured) /stats and the /fastsync/cas/ content gateway.
 func (h *Handler) Mux() *http.ServeMux {
@@ -125,9 +147,53 @@ func (h *Handler) Mux() *http.ServeMux {
 	if h.cas != nil {
 		mux.HandleFunc(fastSyncCASPrefix, h.handleFastSyncCAS)
 	}
+	if h.anchors != nil {
+		mux.HandleFunc(fastSyncAnchorsPath, h.handleFastSyncAnchors)
+	}
 	return mux
 }
 
+// handleFastSyncAnchors serves verifiable anchor bundles for a height range as JSON.
+// Each bundle carries the raw tx and a merkle branch; the client recomputes the txid
+// and checks the branch against the block header's merkle root from its OWN verified
+// header chain (the rawTx and branch hashes are hex of the raw wire bytes).
+func (h *Handler) handleFastSyncAnchors(w http.ResponseWriter, r *http.Request) {
+	setGatewayCORS(w)
+	if r.Method == http.MethodOptions {
+		w.WriteHeader(http.StatusNoContent)
+		return
+	}
+	if r.Method != http.MethodGet {
+		writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "methodNotAllowed"})
+		return
+	}
+	from, err1 := strconv.Atoi(r.URL.Query().Get("from"))
+	to, err2 := strconv.Atoi(r.URL.Query().Get("to"))
+	if err1 != nil || err2 != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalidRange", "message": "from and to must be integer heights"})
+		return
+	}
+	bundles, err := h.anchors.AnchorBundles(int32(from), int32(to))
+	if err != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalidRange", "message": err.Error()})
+		return
+	}
+	out := make([]map[string]any, 0, len(bundles))
+	for _, b := range bundles {
+		branch := make([]string, len(b.Branch))
+		for i, hsh := range b.Branch {
+			branch[i] = hex.EncodeToString(hsh[:])
+		}
+		out = append(out, map[string]any{
+			"height":  b.Height,
+			"txIndex": b.TxIndex,
+			"rawTx":   hex.EncodeToString(b.RawTx),
+			"branch":  branch,
+		})
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"anchors": out})
+}
+
 // setGatewayCORS opens the fast-sync gateway to browser clients. The served data is
 // public and self-verifying (anchors prove inclusion against Bitcoin; CAS blobs are
 // content-addressed), so there is nothing to protect with a same-origin policy.

internal/api/resolution_test.go

@@ -1,6 +1,8 @@
 package api
 
 import (
+	"bytes"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -9,7 +11,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
+
 	"github.com/13x-tech/ion-node/internal/cas"
+	"github.com/13x-tech/ion-node/internal/fastsync"
 	"github.com/13x-tech/ion-node/internal/resolve"
 	"github.com/13x-tech/ion-sdk-go/pkg/did"
 )
@@ -359,3 +366,89 @@ func TestFastSyncCASGateway(t *testing.T) {
 		t.Errorf("no-gateway route = %d, want 404", none.StatusCode)
 	}
 }
+
+// fakeAnchorProvider returns its bundles for any valid range; errors on a bad one.
+type fakeAnchorProvider []fastsync.Bundle
+
+func (f fakeAnchorProvider) AnchorBundles(from, to int32) ([]fastsync.Bundle, error) {
+	if from < 0 || to < from {
+		return nil, fmt.Errorf("bad range [%d,%d]", from, to)
+	}
+	return f, nil
+}
+
+// TestFastSyncAnchorsEndpoint verifies GET /fastsync/anchors serializes bundles such
+// that a client can reconstruct and VERIFY them after the JSON round-trip (the whole
+// point of the feed), plus CORS, bad-range 400, and route-absent-without-provider.
+func TestFastSyncAnchorsEndpoint(t *testing.T) {
+	// A single-tx "block": the branch is empty and the merkle root is the txid, so a
+	// client verifies the reconstructed bundle against that root.
+	mt := wire.NewMsgTx(1)
+	mt.AddTxIn(&wire.TxIn{PreviousOutPoint: wire.OutPoint{Index: 0}, SignatureScript: []byte{0x51}})
+	mt.AddTxOut(wire.NewTxOut(1, []byte{txscript.OP_TRUE}))
+	var raw bytes.Buffer
+	if err := mt.SerializeNoWitness(&raw); err != nil {
+		t.Fatal(err)
+	}
+	txid := mt.TxHash()
+	prov := fakeAnchorProvider{{Height: 700001, TxIndex: 0, RawTx: raw.Bytes(), Branch: nil}}
+	srv := httptest.NewServer(NewHandler(fakeResolver{}, nil).WithAnchorBundles(prov).Mux())
+	defer srv.Close()
+
+	resp, err := http.Get(srv.URL + "/fastsync/anchors?from=700001&to=700001")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("GET anchors = %d, want 200", resp.StatusCode)
+	}
+	if resp.Header.Get("Access-Control-Allow-Origin") != "*" {
+		t.Error("missing CORS header")
+	}
+	var body struct {
+		Anchors []struct {
+			Height  int32    `json:"height"`
+			TxIndex int      `json:"txIndex"`
+			RawTx   string   `json:"rawTx"`
+			Branch  []string `json:"branch"`
+		} `json:"anchors"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+	if len(body.Anchors) != 1 {
+		t.Fatalf("got %d anchors, want 1", len(body.Anchors))
+	}
+	a := body.Anchors[0]
+
+	// Reconstruct the bundle from the JSON and verify it — exactly what a client does.
+	rawTx, err := hex.DecodeString(a.RawTx)
+	if err != nil {
+		t.Fatalf("rawTx not hex: %v", err)
+	}
+	branch := make([]chainhash.Hash, len(a.Branch))
+	for i, hx := range a.Branch {
+		b, _ := hex.DecodeString(hx)
+		copy(branch[i][:], b)
+	}
+	got := fastsync.Bundle{Height: a.Height, TxIndex: a.TxIndex, RawTx: rawTx, Branch: branch}
+	if vtxid, ok := got.Verify(txid); !ok || vtxid != txid {
+		t.Error("reconstructed bundle failed verification after the JSON round-trip")
+	}
+
+	// Bad range → 400.
+	bad, _ := http.Get(srv.URL + "/fastsync/anchors?from=abc&to=10")
+	bad.Body.Close()
+	if bad.StatusCode != http.StatusBadRequest {
+		t.Errorf("bad range = %d, want 400", bad.StatusCode)
+	}
+	// No provider → route absent (404).
+	plain := httptest.NewServer(NewHandler(fakeResolver{}, nil).Mux())
+	defer plain.Close()
+	none, _ := http.Get(plain.URL + "/fastsync/anchors?from=1&to=2")
+	none.Body.Close()
+	if none.StatusCode != http.StatusNotFound {
+		t.Errorf("no-provider route = %d, want 404", none.StatusCode)
+	}
+}

internal/fastsync/service.go

@@ -0,0 +1,51 @@
+package fastsync
+
+import "fmt"
+
+const (
+	// defaultMaxAnchorHeights bounds a single AnchorBundles request's height span.
+	defaultMaxAnchorHeights = 5000
+	// maxBundlesPerResponse bounds a single response's bundle count.
+	maxBundlesPerResponse = 5000
+)
+
+// Service is the transport-agnostic fast-sync serving core (epic #111). Both the
+// HTTP gateway (#115) and the native P2P transport (#122) call into it, so the
+// bounds and trust posture live in one place. It reads verifiable anchor bundles
+// from the store; the caller (client) verifies each against its OWN PoW header chain.
+type Service struct {
+	store      *Store
+	maxHeights int32
+}
+
+// NewService builds a fast-sync service over the given bundle store.
+func NewService(store *Store) *Service {
+	return &Service{store: store, maxHeights: defaultMaxAnchorHeights}
+}
+
+// AnchorBundles returns the verifiable bundles whose confirming height is in
+// [from, to], clamped to the configured max height span and a max bundle count so a
+// single request can't pull an unbounded amount. The bundles are ordered by
+// transaction number (height then tx index).
+func (s *Service) AnchorBundles(from, to int32) ([]Bundle, error) {
+	if from < 0 {
+		return nil, fmt.Errorf("fastsync: negative from height %d", from)
+	}
+	if to < from {
+		return nil, fmt.Errorf("fastsync: to height %d below from height %d", to, from)
+	}
+	if to-from > s.maxHeights {
+		to = from + s.maxHeights
+	}
+	fromTxnum := uint64(uint32(from)) << 32
+	toTxnum := uint64(uint32(to))<<32 | 0xffffffff
+	out := make([]Bundle, 0, 64)
+	err := s.store.Range(fromTxnum, toTxnum, func(_ uint64, b Bundle) bool {
+		out = append(out, b)
+		return len(out) < maxBundlesPerResponse
+	})
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}