universe: add cursor-based delta sync to federation

bench/fixture/discovery.go

@@ -0,0 +1,184 @@
+package fixture
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	"github.com/lightninglabs/taproot-assets/universe"
+)
+
+const (
+	// approxRootWireBytes approximates the wire size of one enumerated
+	// universe root: 32-byte asset ID (or group key hash), 1-byte proof
+	// type, 32-byte node hash, 8-byte sum.
+	approxRootWireBytes = 73
+
+	// approxLeafKeyWireBytes approximates the wire size of one
+	// enumerated leaf key: 36-byte outpoint plus 33-byte script key.
+	approxLeafKeyWireBytes = 69
+)
+
+// DiscoveryMetrics counts the remote DiffEngine traffic generated by a
+// sync run, split by discovery phase. The fixture is in-process, but in
+// production every DiffEngine call is an RPC round trip (or one page
+// thereof), so the call counters approximate round trips and the byte
+// counters approximate wire bytes.
+type DiscoveryMetrics struct {
+	// RootPages counts RootNodes calls (one per enumeration page).
+	RootPages atomic.Int64
+
+	// RootsEnumerated counts roots returned across all RootNodes pages.
+	RootsEnumerated atomic.Int64
+
+	// RootLookups counts RootNode point lookups.
+	RootLookups atomic.Int64
+
+	// LeafKeyPages counts UniverseLeafKeys calls (one per page).
+	LeafKeyPages atomic.Int64
+
+	// LeafKeysEnumerated counts leaf keys returned across all pages.
+	LeafKeysEnumerated atomic.Int64
+
+	// ProofFetches counts FetchProofLeaf calls.
+	ProofFetches atomic.Int64
+
+	// ProofBytes sums the raw proof blob sizes returned by
+	// FetchProofLeaf and by delta pages. A lower bound on proof wire
+	// bytes (inclusion proof overhead is excluded).
+	ProofBytes atomic.Int64
+
+	// DeltaPages counts SyncDelta calls (one per page round trip).
+	DeltaPages atomic.Int64
+
+	// DeltaItems counts leaves returned across all delta pages.
+	DeltaItems atomic.Int64
+}
+
+// Report emits the counters, the derived per-phase byte estimates, and
+// the discovery share of total bytes via b.ReportMetric.
+func (m *DiscoveryMetrics) Report(b *testing.B) {
+	b.Helper()
+
+	rootBytes := m.RootsEnumerated.Load() * approxRootWireBytes
+	keyBytes := m.LeafKeysEnumerated.Load() * approxLeafKeyWireBytes
+	proofBytes := m.ProofBytes.Load()
+
+	discoveryBytes := rootBytes + keyBytes
+	share := float64(0)
+	if total := discoveryBytes + proofBytes; total > 0 {
+		share = float64(discoveryBytes) / float64(total)
+	}
+
+	b.ReportMetric(float64(m.RootPages.Load()), "root_pages")
+	b.ReportMetric(float64(m.RootsEnumerated.Load()), "roots_enum")
+	b.ReportMetric(float64(m.RootLookups.Load()), "root_lookups")
+	b.ReportMetric(float64(m.LeafKeyPages.Load()), "leafkey_pages")
+	b.ReportMetric(float64(m.LeafKeysEnumerated.Load()), "leafkeys_enum")
+	b.ReportMetric(float64(m.ProofFetches.Load()), "proof_fetches")
+	b.ReportMetric(float64(m.DeltaPages.Load()), "delta_pages")
+	b.ReportMetric(float64(m.DeltaItems.Load()), "delta_items")
+	b.ReportMetric(float64(rootBytes), "root_bytes")
+	b.ReportMetric(float64(keyBytes), "leafkey_bytes")
+	b.ReportMetric(float64(proofBytes), "proof_bytes")
+	b.ReportMetric(share, "discovery_share")
+}
+
+// countingDiffEngine wraps a universe.DiffEngine and tallies each call
+// into a DiscoveryMetrics. It is installed around the fixture's remote
+// archive so only remote-side (would-be wire) traffic is counted; the
+// syncer's local enumeration is a DB scan, not a round trip.
+type countingDiffEngine struct {
+	inner   universe.DiffEngine
+	metrics *DiscoveryMetrics
+}
+
+var _ universe.DiffEngine = (*countingDiffEngine)(nil)
+
+// RootNode delegates to the wrapped engine, counting the lookup.
+//
+// NOTE: part of the universe.DiffEngine interface.
+func (c *countingDiffEngine) RootNode(ctx context.Context,
+	id universe.Identifier) (universe.Root, error) {
+
+	c.metrics.RootLookups.Add(1)
+	return c.inner.RootNode(ctx, id)
+}
+
+// RootNodes delegates to the wrapped engine, tallying the page and the
+// roots it enumerates.
+//
+// NOTE: part of the universe.DiffEngine interface.
+func (c *countingDiffEngine) RootNodes(ctx context.Context,
+	q universe.RootNodesQuery) ([]universe.Root, error) {
+
+	roots, err := c.inner.RootNodes(ctx, q)
+	c.metrics.RootPages.Add(1)
+	c.metrics.RootsEnumerated.Add(int64(len(roots)))
+	return roots, err
+}
+
+// UniverseLeafKeys delegates to the wrapped engine, tallying the page
+// and the leaf entries it enumerates.
+//
+// NOTE: part of the universe.DiffEngine interface.
+func (c *countingDiffEngine) UniverseLeafKeys(ctx context.Context,
+	q universe.UniverseLeafKeysQuery) ([]universe.LeafEntry, error) {
+
+	keys, err := c.inner.UniverseLeafKeys(ctx, q)
+	c.metrics.LeafKeyPages.Add(1)
+	c.metrics.LeafKeysEnumerated.Add(int64(len(keys)))
+	return keys, err
+}
+
+// FetchProofLeaf delegates to the wrapped engine, counting the fetch
+// and the proof bytes it returns.
+//
+// NOTE: part of the universe.DiffEngine interface.
+func (c *countingDiffEngine) FetchProofLeaf(ctx context.Context,
+	id universe.Identifier,
+	key universe.LeafKey) ([]*universe.Proof, error) {
+
+	proofs, err := c.inner.FetchProofLeaf(ctx, id, key)
+	c.metrics.ProofFetches.Add(1)
+	for _, p := range proofs {
+		if p.Leaf != nil {
+			c.metrics.ProofBytes.Add(int64(len(p.Leaf.RawProof)))
+		}
+	}
+	return proofs, err
+}
+
+// SyncDelta delegates to the wrapped engine's delta implementation,
+// tallying pages, items, and payload bytes.
+//
+// NOTE: part of the universe.DeltaEngine interface.
+func (c *countingDiffEngine) SyncDelta(ctx context.Context,
+	sinceSeq uint64, pageSize int32) (*universe.DeltaPage, error) {
+
+	deltaEngine, ok := c.inner.(universe.DeltaEngine)
+	if !ok {
+		return nil, universe.ErrDeltaUnsupported
+	}
+
+	page, err := deltaEngine.SyncDelta(ctx, sinceSeq, pageSize)
+	c.metrics.DeltaPages.Add(1)
+	if page != nil {
+		c.metrics.DeltaItems.Add(int64(len(page.Items)))
+		for _, item := range page.Items {
+			if item.Leaf != nil {
+				c.metrics.ProofBytes.Add(
+					int64(len(item.Leaf.RawProof)),
+				)
+			}
+		}
+	}
+	return page, err
+}
+
+// Close closes the wrapped engine.
+//
+// NOTE: part of the universe.DiffEngine interface.
+func (c *countingDiffEngine) Close() error {
+	return c.inner.Close()
+}

bench/fixture/sync.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"math"
 	"math/rand"
 	"strings"
 	"sync/atomic"
@@ -104,6 +105,13 @@ type SeedSpec struct {
 	// the local side starts empty; one means the two sides are already
 	// identical (and sync should be a no-op).
 	LocalOverlap Fraction
+
+	// DivergentRoots, when non-zero, limits the LocalOverlap shortfall
+	// to the first DivergentRoots roots of each sweep; the remaining
+	// roots are seeded fully overlapped (locally identical). This
+	// models a steady-state federation where most universes are quiet
+	// and only a few gained new leaves since the last tick.
+	DivergentRoots int
 }
 
 // universePair is one side of the sync fixture. Each side has its own
@@ -253,6 +261,16 @@ type SyncFixture struct {
 	Remote  *universePair
 	Syncer  *universe.SimpleSyncer
 	Metrics *SyncMetrics
+
+	// Discovery tallies the remote DiffEngine traffic (round trips and
+	// approximate wire bytes) the syncer generates while finding out
+	// what to fetch.
+	Discovery *DiscoveryMetrics
+
+	// seededGens records the asset genesis used for each seeded root,
+	// keyed by proof type, in creation order. This is what allows a
+	// bench to grow a specific already-seeded universe after the fact.
+	seededGens map[universe.ProofType][]asset.Genesis
 }
 
 // NewSyncFixture constructs an unseeded SyncFixture. Call Seed to
@@ -270,6 +288,7 @@ func NewSyncFixture(tb testing.TB, opts SyncFixtureOpts) *SyncFixture {
 	remote := newUniversePair(tb, clk)
 
 	metrics := &SyncMetrics{}
+	discovery := &DiscoveryMetrics{}
 
 	syncer := universe.NewSimpleSyncer(universe.SimpleSyncCfg{
 		LocalDiffEngine: local.Archive,
@@ -280,17 +299,24 @@ func NewSyncFixture(tb testing.TB, opts SyncFixtureOpts) *SyncFixture {
 		NewRemoteDiffEngine: func(
 			_ universe.ServerAddr) (universe.DiffEngine, error) {
 
-			return remote.Archive, nil
+			return &countingDiffEngine{
+				inner:   remote.Archive,
+				metrics: discovery,
+			}, nil
 		},
 		SyncBatchSize:       opts.SyncBatchSize,
 		SyncRootConcurrency: opts.SyncRootConcurrency,
 	})
 
 	return &SyncFixture{
-		Local:   local,
-		Remote:  remote,
-		Syncer:  syncer,
-		Metrics: metrics,
+		Local:     local,
+		Remote:    remote,
+		Syncer:    syncer,
+		Metrics:   metrics,
+		Discovery: discovery,
+		seededGens: make(
+			map[universe.ProofType][]asset.Genesis,
+		),
 	}
 }
 
@@ -324,14 +350,15 @@ func (f *SyncFixture) Seed(tb testing.TB, spec SeedSpec) {
 	ctx := context.Background()
 
 	seedType(tb, ctx, f, universe.ProofTypeIssuance, spec.Issuance,
-		spec.LocalOverlap)
+		spec.LocalOverlap, spec.DivergentRoots)
 	seedType(tb, ctx, f, universe.ProofTypeTransfer, spec.Transfer,
-		spec.LocalOverlap)
+		spec.LocalOverlap, spec.DivergentRoots)
 }
 
 // seedType is the per-proof-type worker used by Seed.
 func seedType(tb testing.TB, ctx context.Context, f *SyncFixture,
-	pt universe.ProofType, sweep RootSweep, overlap Fraction) {
+	pt universe.ProofType, sweep RootSweep, overlap Fraction,
+	divergentRoots int) {
 
 	tb.Helper()
 
@@ -342,6 +369,12 @@ func seedType(tb testing.TB, ctx context.Context, f *SyncFixture,
 	localCount := int(float64(sweep.Leaves) * float64(overlap))
 
 	for r := 0; r < sweep.Roots; r++ {
+		// Roots beyond the divergent prefix are seeded fully
+		// overlapped: the local side already has every leaf.
+		rootLocalCount := localCount
+		if divergentRoots > 0 && r >= divergentRoots {
+			rootLocalCount = sweep.Leaves
+		}
 		// All leaves under one root share a single asset genesis; the
 		// universe identifier is derived from that same genesis so
 		// insert-time and query-time namespaces agree. Deriving id
@@ -353,6 +386,7 @@ func seedType(tb testing.TB, ctx context.Context, f *SyncFixture,
 			AssetID:   assetGen.ID(),
 			ProofType: pt,
 		}
+		f.seededGens[pt] = append(f.seededGens[pt], assetGen)
 
 		remoteItems := make([]*universe.Item, sweep.Leaves)
 		for i := 0; i < sweep.Leaves; i++ {
@@ -370,17 +404,58 @@ func seedType(tb testing.TB, ctx context.Context, f *SyncFixture,
 		)
 		require.NoError(tb, err)
 
-		if localCount == 0 {
+		if rootLocalCount == 0 {
 			continue
 		}
 
 		err = f.Local.Multiverse.UpsertProofLeafBatch(
-			ctx, remoteItems[:localCount],
+			ctx, remoteItems[:rootLocalCount],
 		)
 		require.NoError(tb, err)
 	}
 }
 
+// AddRemoteDivergence grows an already-seeded universe on the remote
+// side only: n new leaves are appended to the rootIdx-th root of the
+// given proof type's sweep. This models the steady-state shape where a
+// previously synced universe gains new leaves between federation ticks.
+func (f *SyncFixture) AddRemoteDivergence(tb testing.TB,
+	pt universe.ProofType, rootIdx, n int) {
+
+	tb.Helper()
+
+	gens := f.seededGens[pt]
+	require.Less(tb, rootIdx, len(gens), "root index out of range")
+
+	assetGen := gens[rootIdx]
+	id := universe.Identifier{
+		AssetID:   assetGen.ID(),
+		ProofType: pt,
+	}
+
+	ctx := context.Background()
+	for i := 0; i < n; i++ {
+		_, err := f.Remote.Multiverse.UpsertProofLeaf(
+			ctx, id, randLeafKey(tb),
+			randMintingLeafFor(tb, assetGen), nil,
+		)
+		require.NoError(tb, err)
+	}
+}
+
+// RemoteMaxSeq returns the remote side's current insertion-journal
+// high-water mark, i.e. the cursor a fully synced peer would hold.
+func (f *SyncFixture) RemoteMaxSeq(tb testing.TB) uint64 {
+	tb.Helper()
+
+	_, maxSeq, err := f.Remote.Multiverse.FetchLeavesSince(
+		context.Background(), 0, int32(math.MaxInt32),
+	)
+	require.NoError(tb, err)
+
+	return maxSeq
+}
+
 // randLeafKey returns a random universe leaf key. Each call allocates
 // a fresh *asset.ScriptKey, which is exactly the shape that reveals
 // the pointer-identity diff bug when the returned keys are diffed via