diff --git a/docs/release-notes/release-notes-0.8.2.md b/docs/release-notes/release-notes-0.8.2.md index 347000bd7..919aa7148 100644 --- a/docs/release-notes/release-notes-0.8.2.md +++ b/docs/release-notes/release-notes-0.8.2.md @@ -45,11 +45,14 @@ ## Performance Improvements +- [PR#2183](https://github.com/lightninglabs/taproot-assets/pull/2183) + dramatically improves the performance of MS-SMT proof verification. + - [PR#2184](https://github.com/lightninglabs/taproot-assets/pull/2184) dramatically improves the performance of universe federation proof push. -- [PR#2183](https://github.com/lightninglabs/taproot-assets/pull/2183) - dramatically improves the performance of MS-SMT proof verification. +- [PR#2194](https://github.com/lightninglabs/taproot-assets/pull/2194) + significantly improves concurrent universe proof ingest on Postgres. ## Deprecations diff --git a/mssmt/tree_prop_test.go b/mssmt/tree_prop_test.go new file mode 100644 index 000000000..828542381 --- /dev/null +++ b/mssmt/tree_prop_test.go @@ -0,0 +1,99 @@ +package mssmt_test + +import ( + "context" + "testing" + + "github.com/lightninglabs/taproot-assets/mssmt" + "github.com/stretchr/testify/require" + "pgregory.net/rapid" +) + +// testInsertLastWriteWins asserts that insertion is last-write-wins per +// key: applying a sequence of inserts that reuses keys produces the +// same tree as inserting only the final leaf observed for each key. +// This is the invariant that allows coalescing consecutive updates to +// the same key into a single insert of the latest value. +func testInsertLastWriteWins(t *rapid.T) { + ctx := context.Background() + + // Draw a small pool of keys so the insertion sequence is likely + // to hit the same key several times. Duplicate draws within the + // pool are harmless; they only shrink the effective pool. + numKeys := rapid.IntRange(1, 8).Draw(t, "num_keys") + keys := make([][hashSize]byte, numKeys) + for i := range keys { + keyBytes := rapid.SliceOfN( + rapid.Byte(), hashSize, hashSize, + ).Draw(t, "key") + copy(keys[i][:], keyBytes) + } + + // Draw the insertion sequence. Sums are bounded well below the + // point where the tree's uint64 sum overflow check could + // trigger. + numInserts := rapid.IntRange(1, 32).Draw(t, "num_inserts") + sequence := make([]treeLeaf, numInserts) + for i := range sequence { + keyIdx := rapid.IntRange(0, numKeys-1).Draw(t, "key_idx") + value := rapid.SliceOfN(rapid.Byte(), 1, 64).Draw(t, "value") + sum := rapid.Uint64Range(0, 1<<32).Draw(t, "sum") + + sequence[i] = treeLeaf{ + key: keys[keyIdx], + leaf: mssmt.NewLeafNode(value, sum), + } + } + + // Apply the full sequence in order. + full := mssmt.NewCompactedTree(mssmt.NewDefaultStore()) + for _, item := range sequence { + _, err := full.Insert(ctx, item.key, item.leaf) + require.NoError(t, err) + } + + // Reduce the sequence to the final leaf per key, keeping the + // order of each key's first occurrence, and apply only those. + finalLeaves := make(map[[hashSize]byte]*mssmt.LeafNode) + var keyOrder [][hashSize]byte + for _, item := range sequence { + if _, ok := finalLeaves[item.key]; !ok { + keyOrder = append(keyOrder, item.key) + } + finalLeaves[item.key] = item.leaf + } + + coalesced := mssmt.NewCompactedTree(mssmt.NewDefaultStore()) + for _, key := range keyOrder { + _, err := coalesced.Insert(ctx, key, finalLeaves[key]) + require.NoError(t, err) + } + + fullRoot, err := full.Root(ctx) + require.NoError(t, err) + coalescedRoot, err := coalesced.Root(ctx) + require.NoError(t, err) + + require.True( + t, mssmt.IsEqualNode(fullRoot, coalescedRoot), + "full root %v != coalesced root %v", fullRoot, coalescedRoot, + ) + + // Each key's final leaf must carry a valid inclusion proof in + // the fully-inserted tree. + for _, key := range keyOrder { + proof, err := full.MerkleProof(ctx, key) + require.NoError(t, err) + require.True(t, mssmt.VerifyMerkleProof( + key, finalLeaves[key], proof, fullRoot, + )) + } +} + +// TestInsertLastWriteWins runs the last-write-wins insertion property +// against the compacted tree. +func TestInsertLastWriteWins(t *testing.T) { + t.Parallel() + + rapid.Check(t, testInsertLastWriteWins) +} diff --git a/tapcfg/server.go b/tapcfg/server.go index 87c676f9b..c2cdc4657 100644 --- a/tapcfg/server.go +++ b/tapcfg/server.go @@ -193,6 +193,15 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger, return nil, fmt.Errorf("create multiverse store: %w", err) } + // The multiverse trees are derived from the universe roots; repair any + // entries that diverged, for example because the daemon stopped + // between a proof insert committing and its multiverse update being + // written. + err = multiverse.ReconcileMultiverse(context.Background()) + if err != nil { + return nil, fmt.Errorf("reconcile multiverse: %w", err) + } + uniStatsDB := tapdb.NewTransactionExecutor( db, func(tx *sql.Tx) tapdb.UniverseStatsStore { return db.WithTx(tx) diff --git a/tapdb/interfaces.go b/tapdb/interfaces.go index d4dd9f421..02bdcc2fa 100644 --- a/tapdb/interfaces.go +++ b/tapdb/interfaces.go @@ -322,13 +322,30 @@ type BaseDB struct { *sqlc.Queries } +// TxIsolationOverrider is an optional interface that TxOptions can implement +// to override the default serializable isolation level a transaction is +// started with. The override is only honored on the Postgres backend; +// SQLite's driver only supports its default isolation. +type TxIsolationOverrider interface { + // TxIsolation returns the isolation level to start the transaction + // with. + TxIsolation() sql.IsolationLevel +} + // BeginTx wraps the normal sql specific BeginTx method with the TxOptions // interface. This interface is then mapped to the concrete sql tx options // struct. func (s *BaseDB) BeginTx(ctx context.Context, opts TxOptions) (*sql.Tx, error) { + isolation := sql.LevelSerializable + if o, ok := opts.(TxIsolationOverrider); ok && + s.Backend() == sqlc.BackendTypePostgres { + + isolation = o.TxIsolation() + } + sqlOptions := sql.TxOptions{ ReadOnly: opts.ReadOnly(), - Isolation: sql.LevelSerializable, + Isolation: isolation, } return s.DB.BeginTx(ctx, &sqlOptions) } diff --git a/tapdb/multiverse.go b/tapdb/multiverse.go index 0176c37e9..273bf5b54 100644 --- a/tapdb/multiverse.go +++ b/tapdb/multiverse.go @@ -6,6 +6,7 @@ import ( "database/sql" "errors" "fmt" + "sync" "time" "github.com/btcsuite/btcd/btcec/v2" @@ -132,6 +133,17 @@ type MultiverseStore struct { leafKeysCache *universeLeafPageCache + // rootCoalescer batches all writes to the shared multiverse + // trees, so proof insert transactions never touch rows that are + // contended across universes. + rootCoalescer *multiverseRootCoalescer + + // multiverseWriteMu serializes all multiverse writes in this + // process: the coalescer's flushes and the deletion paths. This + // mutual exclusion is what makes it safe for flushes to run + // below serializable isolation. + multiverseWriteMu sync.Mutex + // transferProofDistributor is an event distributor that will be used to // notify subscribers about new proof leaves that are added to the // multiverse. This is used to notify the custodian about new incoming @@ -149,7 +161,7 @@ func NewMultiverseStore(db BatchedMultiverse, return nil, fmt.Errorf("parse max proof cache size: %w", err) } - return &MultiverseStore{ + store := &MultiverseStore{ db: db, cfg: cfg, syncerCache: newSyncerRootNodeCache( @@ -165,7 +177,12 @@ func NewMultiverseStore(db BatchedMultiverse, cfg.Caches.LeavesPerUniverse, ), transferProofDistributor: fn.NewEventDistributor[proof.Blob](), - }, nil + } + store.rootCoalescer = newMultiverseRootCoalescer( + db, &store.multiverseWriteMu, + ) + + return store, nil } // namespaceForProof returns the multiverse namespace used for the given proof @@ -790,10 +807,8 @@ func (b *MultiverseStore) UpsertProofLeaf(ctx context.Context, metaReveal *proof.MetaReveal) (*universe.Proof, error) { var ( - writeTx BaseMultiverseOptions - uniProof *universe.Proof - multiverseRoot mssmt.Node - multiverseProof *mssmt.Proof + writeTx BaseMultiverseOptions + uniProof *universe.Proof ) execTxFunc := func(dbTx BaseMultiverseStore) error { @@ -813,17 +828,6 @@ func (b *MultiverseStore) UpsertProofLeaf(ctx context.Context, return fmt.Errorf("failed universe upsert: %w", err) } - // Now, attempt to insert the universe root into the main - // multiverse tree. - // - // nolint:lll - multiverseRoot, multiverseProof, err = upsertMultiverseLeafEntry( - ctx, dbTx, id, uniProof.UniverseRoot, - ) - if err != nil { - return fmt.Errorf("failed multiverse upsert: %w", err) - } - return nil } dbErr := b.db.ExecTx(ctx, &writeTx, execTxFunc) @@ -831,8 +835,22 @@ func (b *MultiverseStore) UpsertProofLeaf(ctx context.Context, return nil, dbErr } + // Now reflect the universe's new root in the shared multiverse tree, + // through the root coalescer rather than the transaction above: the + // multiverse rows are contended by every insert into every universe, + // so writing them under the insert's own transaction would serialize + // ingest across universes. If this fails, the universe leaf above + // remains committed, and the universe's multiverse entry is healed by + // its next successful update. + multiverseRoot, multiverseProof, err := b.rootCoalescer.updateRoot( + ctx, id, uniProof.UniverseRoot, + ) + if err != nil { + return nil, fmt.Errorf("failed multiverse upsert: %w", err) + } + // Populate the multiverse fields in the proof object now that the - // transaction is complete. + // update is complete. uniProof.MultiverseRoot = multiverseRoot uniProof.MultiverseInclusionProof = multiverseProof @@ -872,9 +890,24 @@ func (b *MultiverseStore) UpsertProofLeafBatch(ctx context.Context, writeTx BaseMultiverseOptions uniProofs []*universe.Proof ) + // Track the final universe root per universe, so the shared + // multiverse tree is updated once per universe with its latest root, + // rather than once per item. + var ( + finalRoots map[universeIDKey]universeRootUpdate + updateOrder []universeIDKey + ) + dbErr := b.db.ExecTx( ctx, &writeTx, func(store BaseMultiverseStore) error { uniProofs = make([]*universe.Proof, len(items)) + + finalRoots = make( + map[universeIDKey]universeRootUpdate, + len(items), + ) + updateOrder = nil + for idx := range items { item := items[idx] @@ -902,24 +935,14 @@ func (b *MultiverseStore) UpsertProofLeafBatch(ctx context.Context, } uniProofs[idx] = uniProof - // Next we'll, attempt to insert the universe - // root into the main multiverse tree. - // - //nolint:lll - multiRoot, multiProof, err := upsertMultiverseLeafEntry( - ctx, store, item.ID, - uniProof.UniverseRoot, - ) - if err != nil { - return fmt.Errorf("failed multiverse "+ - "upsert for item %d: %w", - idx, err) + key := item.ID.String() + if _, ok := finalRoots[key]; !ok { + updateOrder = append(updateOrder, key) + } + finalRoots[key] = universeRootUpdate{ + id: item.ID, + root: uniProof.UniverseRoot, } - - // Update the proof object with multiverse - // details. - uniProofs[idx].MultiverseRoot = multiRoot - uniProofs[idx].MultiverseInclusionProof = multiProof //nolint:lll } return nil @@ -929,6 +952,22 @@ func (b *MultiverseStore) UpsertProofLeafBatch(ctx context.Context, return dbErr } + // Now reflect each universe's final root in the shared multiverse + // tree, through the root coalescer rather than the transaction above: + // the multiverse rows are contended by every insert into every + // universe, so writing them under the batch's own transaction would + // collide with concurrent inserts. If this fails, the universe leaves + // above remain committed, and each universe's multiverse entry is + // healed by its next successful update. + updates := make([]universeRootUpdate, 0, len(updateOrder)) + for _, key := range updateOrder { + updates = append(updates, finalRoots[key]) + } + err := b.rootCoalescer.updateRoots(ctx, updates) + if err != nil { + return fmt.Errorf("failed multiverse upsert: %w", err) + } + // TODO(roasbeef): want to write thru but then need db query again? b.rootNodeCache.wipeCache() @@ -978,6 +1017,10 @@ func (b *MultiverseStore) DeleteUniverse(ctx context.Context, var writeTx BaseUniverseStoreOptions + // Deleting touches the shared multiverse tree, so take the + // multiverse write lock to stay mutually exclusive with the root + // coalescer's flushes. + b.multiverseWriteMu.Lock() dbErr := b.db.ExecTx(ctx, &writeTx, func(tx BaseMultiverseStore) error { multiverseNS, err := namespaceForProof(id.ProofType) if err != nil { @@ -996,6 +1039,7 @@ func (b *MultiverseStore) DeleteUniverse(ctx context.Context, return deleteUniverseTree(ctx, tx, id) }) + b.multiverseWriteMu.Unlock() if dbErr != nil { return "", dbErr } @@ -1019,6 +1063,12 @@ func (b *MultiverseStore) DeleteProofLeaf(ctx context.Context, var writeTx BaseMultiverseOptions + // Deleting touches the shared multiverse tree, so take the + // multiverse write lock to stay mutually exclusive with the root + // coalescer's flushes. + b.multiverseWriteMu.Lock() + defer b.multiverseWriteMu.Unlock() + dbErr := b.db.ExecTx( ctx, &writeTx, func(tx BaseMultiverseStore) error { namespace := id.String() @@ -1060,7 +1110,7 @@ func (b *MultiverseStore) DeleteProofLeaf(ctx context.Context, // Otherwise, update the multiverse entry with the // new universe root. - _, _, err = upsertMultiverseLeafEntry( + err = upsertMultiverseLeafEntry( ctx, tx, id, newRoot, ) if err != nil { diff --git a/tapdb/multiverse_coalescer.go b/tapdb/multiverse_coalescer.go new file mode 100644 index 000000000..ba44354d5 --- /dev/null +++ b/tapdb/multiverse_coalescer.go @@ -0,0 +1,377 @@ +package tapdb + +import ( + "context" + "database/sql" + "fmt" + "runtime/debug" + "sync" + + "github.com/lightninglabs/taproot-assets/mssmt" + "github.com/lightninglabs/taproot-assets/universe" +) + +// multiverseFlushTx is the transaction options for the coalescer's +// flushes. The flush is the sole writer of the multiverse namespaces +// (enforced by the single-flusher role plus the multiverse write mutex +// shared with the deletion paths), so serializable isolation buys it +// nothing. Running it at read committed on Postgres means it takes no +// predicate locks, and, since only serializable writers flag conflicts +// on serializable readers, it can neither abort nor be aborted by the +// concurrent universe insert transactions. +type multiverseFlushTx struct{} + +// ReadOnly returns false: flushes write. +func (multiverseFlushTx) ReadOnly() bool { + return false +} + +// TxIsolation overrides the flush transaction's isolation level. +func (multiverseFlushTx) TxIsolation() sql.IsolationLevel { + return sql.LevelReadCommitted +} + +// multiverseRootUpdate is the outcome of a flushed multiverse root +// update: the multiverse root after the flush that carried the update, +// along with the inclusion proof for the universe's leaf within that +// root. +type multiverseRootUpdate struct { + // multiverseRoot is the root of the multiverse tree after the + // flush. + multiverseRoot mssmt.Node + + // inclusionProof is the inclusion proof of the universe's leaf + // within multiverseRoot. + inclusionProof *mssmt.Proof +} + +// flushResult is what a waiter receives once the flush carrying its +// update has completed, or failed. +type flushResult struct { + update multiverseRootUpdate + err error +} + +// flushWaiter is a caller awaiting the flush of a pending update. Only +// waiters that want the multiverse root and inclusion proof cause them +// to be generated; batch callers skip that work. +type flushWaiter struct { + result chan flushResult + wantProof bool +} + +// universeRootUpdate names a universe's new root for submission to the +// coalescer. +type universeRootUpdate struct { + id universe.Identifier + root mssmt.Node +} + +// pendingRootUpdate is a universe's pending multiverse root update. It +// holds the latest universe root submitted for the universe, along +// with every caller awaiting a flush that carries it. Keeping at most +// one pending update per universe is what coalesces redundant +// multiverse writes: SMT insertion is last-write-wins per key, so only +// the latest root needs to be written. +type pendingRootUpdate struct { + id universe.Identifier + root mssmt.Node + waiters []flushWaiter +} + +// multiverseRootCoalescer serializes all writes to the shared +// multiverse trees through a single flusher, coalescing concurrent +// updates into one write transaction. +// +// Every proof leaf insert must reflect its universe's new root in the +// shared multiverse tree for its proof type. Doing that write inside +// each insert's own transaction makes any two concurrent inserts +// collide on the multiverse root rows: under Postgres serializable +// isolation one of them aborts and retries with backoff, effectively +// serializing ingest across universes. Routing the updates through the +// coalescer removes the shared rows from the insert transactions +// entirely: inserts commit in parallel, and their multiverse updates +// are applied by at most one flusher at a time, batched together. +// +// The flusher role is leader-based rather than a dedicated goroutine: +// the first caller to find the coalescer idle flushes pending updates +// in rounds until none remain, while every other caller just awaits +// its result. This yields group commit without any lifecycle +// management: at low load an update flushes immediately, and under +// load updates accumulate while a flush is in flight and are applied +// together in the next round. +type multiverseRootCoalescer struct { + db BatchedMultiverse + + // writeMu serializes all multiverse writes in this process. The + // flusher holds it for the duration of a flush transaction, and + // the deletion paths hold it around theirs. This mutual + // exclusion is what makes it safe to run flushes below + // serializable isolation. + writeMu sync.Locker + + mu sync.Mutex + + // pending holds the latest submitted universe root per universe. + // The map keying enforces at most one pending update per + // universe. + pending map[universeIDKey]*pendingRootUpdate + + // order tracks the first-submission order of pending universes, + // so flushes apply updates deterministically. + order []universeIDKey + + // flushing is true while some caller holds the flusher role. + flushing bool +} + +// newMultiverseRootCoalescer creates a new coalescer that writes +// through the given db handle. The given lock must be held by every +// other multiverse writer in the process. +func newMultiverseRootCoalescer(db BatchedMultiverse, + writeMu sync.Locker) *multiverseRootCoalescer { + + return &multiverseRootCoalescer{ + db: db, + writeMu: writeMu, + pending: make(map[universeIDKey]*pendingRootUpdate), + } +} + +// updateRoot records the new root of the given universe and returns +// once a flush carrying the update has committed, returning the +// multiverse root and the universe leaf's inclusion proof from that +// flush. +// +// If several updates for the same universe are submitted concurrently, +// they coalesce: the flushed leaf value is the latest submitted root, +// and all of the universe's waiters receive the root and proof of the +// flush that carried it. +func (c *multiverseRootCoalescer) updateRoot(ctx context.Context, + id universe.Identifier, root mssmt.Node) (mssmt.Node, *mssmt.Proof, + error) { + + c.mu.Lock() + result := c.enqueue(id, root, true) + lead := !c.flushing + if lead { + c.flushing = true + } + c.mu.Unlock() + + if lead { + c.flush() + } + + select { + case res := <-result: + if res.err != nil { + return nil, nil, res.err + } + + return res.update.multiverseRoot, res.update.inclusionProof, + nil + + case <-ctx.Done(): + return nil, nil, ctx.Err() + } +} + +// updateRoots records the new roots of the given universes and returns +// once a flush carrying every update has committed. Unlike updateRoot, +// it does not cause multiverse roots or inclusion proofs to be +// generated, as batch callers do not consume them. +func (c *multiverseRootCoalescer) updateRoots(ctx context.Context, + updates []universeRootUpdate) error { + + if len(updates) == 0 { + return nil + } + + waiters := make([]chan flushResult, len(updates)) + + c.mu.Lock() + for i, update := range updates { + waiters[i] = c.enqueue(update.id, update.root, false) + } + lead := !c.flushing + if lead { + c.flushing = true + } + c.mu.Unlock() + + if lead { + c.flush() + } + + for _, waiter := range waiters { + select { + case res := <-waiter: + if res.err != nil { + return res.err + } + + case <-ctx.Done(): + return ctx.Err() + } + } + + return nil +} + +// enqueue records the latest root for the given universe and registers +// a waiter for the flush that carries it. +// +// NOTE: The caller must hold c.mu. +func (c *multiverseRootCoalescer) enqueue(id universe.Identifier, + root mssmt.Node, wantProof bool) chan flushResult { + + result := make(chan flushResult, 1) + + key := id.String() + update, ok := c.pending[key] + if !ok { + update = &pendingRootUpdate{id: id} + c.pending[key] = update + c.order = append(c.order, key) + } + update.root = root + update.waiters = append(update.waiters, flushWaiter{ + result: result, + wantProof: wantProof, + }) + + return result +} + +// flush drains pending updates in rounds until none remain, applying +// each round in a single write transaction. +func (c *multiverseRootCoalescer) flush() { + for { + c.mu.Lock() + if len(c.order) == 0 { + c.flushing = false + c.mu.Unlock() + return + } + + batch := make([]*pendingRootUpdate, 0, len(c.order)) + for _, key := range c.order { + batch = append(batch, c.pending[key]) + } + c.pending = make(map[universeIDKey]*pendingRootUpdate) + c.order = nil + c.mu.Unlock() + + c.flushBatch(batch) + } +} + +// flushBatch applies one round of updates in a single write transaction +// and delivers the outcome to every waiter of the round. +func (c *multiverseRootCoalescer) flushBatch(batch []*pendingRootUpdate) { + // A panic during the flush must not strand the round's waiters or + // unwind past the flusher role's bookkeeping: were the panic + // recovered further up the stack, the coalescer would be blocked + // for every future update. Convert it into an error for every + // waiter of the round instead, and let the flusher continue. + defer func() { + r := recover() + if r == nil { + return + } + + log.Criticalf("Multiverse flush panic: %v\n%s", r, + debug.Stack()) + + res := flushResult{ + err: fmt.Errorf("multiverse flush panic: %v", r), + } + for _, update := range batch { + // The waiter channels are buffered, so a send only + // fails if the waiter was already served by the + // normal delivery path below. + for _, waiter := range update.waiters { + select { + case waiter.result <- res: + default: + } + } + } + }() + + // The flush must not be tied to any single caller's context: the + // universe transactions the updates stem from have already + // committed, and other callers in the round await this write. + ctx := context.Background() + + wantProof := func(update *pendingRootUpdate) bool { + for _, waiter := range update.waiters { + if waiter.wantProof { + return true + } + } + + return false + } + + c.writeMu.Lock() + defer c.writeMu.Unlock() + + results := make([]multiverseRootUpdate, len(batch)) + flushTx := multiverseFlushTx{} + err := c.db.ExecTx( + ctx, flushTx, func(store BaseMultiverseStore) error { + // Write every universe's latest root first, then + // read back the resulting root and one inclusion + // proof per universe that has a waiter consuming + // them. + for _, update := range batch { + err := upsertMultiverseLeafEntry( + ctx, store, update.id, update.root, + ) + if err != nil { + return fmt.Errorf("failed multiverse "+ + "upsert for %v: %w", + update.id.String(), err) + } + } + + for i, update := range batch { + if !wantProof(update) { + continue + } + + root, proof, err := multiverseRootAndProof( + ctx, store, update.id, + ) + if err != nil { + return fmt.Errorf("failed multiverse "+ + "root fetch for %v: %w", + update.id.String(), err) + } + + results[i] = multiverseRootUpdate{ + multiverseRoot: root, + inclusionProof: proof, + } + } + + return nil + }, + ) + + for i, update := range batch { + res := flushResult{err: err} + if err == nil { + res.update = results[i] + } + + // Every waiter channel is buffered, so delivery never + // blocks the flusher, even if a waiter has abandoned its + // call due to context cancellation. + for _, waiter := range update.waiters { + waiter.result <- res + } + } +} diff --git a/tapdb/multiverse_coalescer_test.go b/tapdb/multiverse_coalescer_test.go new file mode 100644 index 000000000..6bea106f9 --- /dev/null +++ b/tapdb/multiverse_coalescer_test.go @@ -0,0 +1,478 @@ +package tapdb + +import ( + "context" + "sync" + "testing" + + "github.com/lightninglabs/taproot-assets/internal/test" + "github.com/lightninglabs/taproot-assets/mssmt" + "github.com/lightninglabs/taproot-assets/universe" + "github.com/stretchr/testify/require" + "pgregory.net/rapid" +) + +// multiverseLeafForRoot builds the multiverse leaf that must commit to +// the given universe root, mirroring the sum rule of the production +// insert path: issuance leaves carry a sum of one, transfer leaves +// carry the universe root's sum. +func multiverseLeafForRoot(id universe.Identifier, + root mssmt.Node) *mssmt.LeafNode { + + rootHash := root.NodeHash() + sum := root.NodeSum() + if id.ProofType == universe.ProofTypeIssuance { + sum = 1 + } + + return mssmt.NewLeafNode(rootHash[:], sum) +} + +// assertOracleRoot asserts that the multiverse root stored for the +// given proof type matches the root of the given in-memory oracle +// tree. +func assertOracleRoot(t require.TestingT, ctx context.Context, + multiverse *MultiverseStore, oracle *mssmt.CompactedTree, + proofType universe.ProofType) { + + oracleRoot, err := oracle.Root(ctx) + require.NoError(t, err) + + storeRoot, err := multiverse.MultiverseRootNode(ctx, proofType) + require.NoError(t, err) + require.True(t, storeRoot.IsSome()) + + storeRoot.WhenSome(func(r universe.MultiverseRoot) { + require.Equal( + t, oracleRoot.NodeHash(), r.Node.NodeHash(), + "proof type %v: root hash mismatch", proofType, + ) + require.Equal( + t, oracleRoot.NodeSum(), r.Node.NodeSum(), + "proof type %v: root sum mismatch", proofType, + ) + }) +} + +// TestMultiverseRootCoalescer asserts that concurrent root updates for +// distinct universes all succeed, return self-verifying inclusion +// proofs, and leave the multiverse trees identical to inserting the +// same leaves directly. +func TestMultiverseRootCoalescer(t *testing.T) { + t.Parallel() + + ctx := context.Background() + multiverse, _ := newTestMultiverse(t) + + const numUniverses = 16 + + type submission struct { + id universe.Identifier + root mssmt.Node + } + subs := make([]submission, numUniverses) + for i := range subs { + proofType := universe.ProofTypeIssuance + if i%2 == 1 { + proofType = universe.ProofTypeTransfer + } + + var id universe.Identifier + id.ProofType = proofType + copy(id.AssetID[:], test.RandBytes(32)) + + var rootHash mssmt.NodeHash + copy(rootHash[:], test.RandBytes(32)) + + subs[i] = submission{ + id: id, + root: mssmt.NewComputedBranch( + rootHash, uint64(test.RandInt[uint32]()), + ), + } + } + + var ( + wg sync.WaitGroup + results = make([]multiverseRootUpdate, numUniverses) + errs = make([]error, numUniverses) + ) + for i := range subs { + wg.Add(1) + go func(i int) { + defer wg.Done() + + root, proof, err := multiverse.rootCoalescer.updateRoot( + ctx, subs[i].id, subs[i].root, + ) + results[i] = multiverseRootUpdate{ + multiverseRoot: root, + inclusionProof: proof, + } + errs[i] = err + }(i) + } + wg.Wait() + + // Every submission must have succeeded and returned an inclusion + // proof for the exact leaf submitted, valid against the returned + // multiverse root. + for i := range subs { + require.NoError(t, errs[i]) + + leaf := multiverseLeafForRoot(subs[i].id, subs[i].root) + require.True(t, mssmt.VerifyMerkleProof( + subs[i].id.Bytes(), leaf, results[i].inclusionProof, + results[i].multiverseRoot, + )) + } + + // The final multiverse roots must equal those of oracle trees + // built directly from the submitted leaves. + oracles := map[universe.ProofType]*mssmt.CompactedTree{ + universe.ProofTypeIssuance: mssmt.NewCompactedTree( + mssmt.NewDefaultStore(), + ), + universe.ProofTypeTransfer: mssmt.NewCompactedTree( + mssmt.NewDefaultStore(), + ), + } + for i := range subs { + id := subs[i].id + leaf := multiverseLeafForRoot(id, subs[i].root) + _, err := oracles[id.ProofType].Insert(ctx, id.Bytes(), leaf) + require.NoError(t, err) + } + for proofType, oracle := range oracles { + assertOracleRoot(t, ctx, multiverse, oracle, proofType) + } +} + +// TestMultiverseRootCoalescerBatch asserts that batch root updates, +// interleaved with concurrent single updates, leave the multiverse +// trees identical to inserting the same leaves directly. +func TestMultiverseRootCoalescerBatch(t *testing.T) { + t.Parallel() + + ctx := context.Background() + multiverse, _ := newTestMultiverse(t) + + newUpdate := func(proofType universe.ProofType) universeRootUpdate { + var id universe.Identifier + id.ProofType = proofType + copy(id.AssetID[:], test.RandBytes(32)) + + var rootHash mssmt.NodeHash + copy(rootHash[:], test.RandBytes(32)) + + return universeRootUpdate{ + id: id, + root: mssmt.NewComputedBranch( + rootHash, uint64(test.RandInt[uint32]()), + ), + } + } + + // One batch of issuance updates, one batch of transfer updates, + // and a handful of single updates, all submitted concurrently. + issuanceBatch := make([]universeRootUpdate, 8) + for i := range issuanceBatch { + issuanceBatch[i] = newUpdate(universe.ProofTypeIssuance) + } + transferBatch := make([]universeRootUpdate, 8) + for i := range transferBatch { + transferBatch[i] = newUpdate(universe.ProofTypeTransfer) + } + singles := make([]universeRootUpdate, 4) + for i := range singles { + singles[i] = newUpdate(universe.ProofTypeIssuance) + } + + var ( + wg sync.WaitGroup + batchErrs = make([]error, 2) + singleErrs = make([]error, len(singles)) + ) + wg.Add(2) + go func() { + defer wg.Done() + batchErrs[0] = multiverse.rootCoalescer.updateRoots( + ctx, issuanceBatch, + ) + }() + go func() { + defer wg.Done() + batchErrs[1] = multiverse.rootCoalescer.updateRoots( + ctx, transferBatch, + ) + }() + for i := range singles { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, _, singleErrs[i] = + multiverse.rootCoalescer.updateRoot( + ctx, singles[i].id, singles[i].root, + ) + }(i) + } + wg.Wait() + + for _, err := range batchErrs { + require.NoError(t, err) + } + for _, err := range singleErrs { + require.NoError(t, err) + } + + oracles := map[universe.ProofType]*mssmt.CompactedTree{ + universe.ProofTypeIssuance: mssmt.NewCompactedTree( + mssmt.NewDefaultStore(), + ), + universe.ProofTypeTransfer: mssmt.NewCompactedTree( + mssmt.NewDefaultStore(), + ), + } + allUpdates := append( + append(issuanceBatch, transferBatch...), singles..., + ) + for _, update := range allUpdates { + _, err := oracles[update.id.ProofType].Insert( + ctx, update.id.Bytes(), + multiverseLeafForRoot(update.id, update.root), + ) + require.NoError(t, err) + } + for proofType, oracle := range oracles { + assertOracleRoot(t, ctx, multiverse, oracle, proofType) + } +} + +// TestMultiverseRootCoalescerProps property-tests the coalescer against +// an in-memory oracle: for any schedule of concurrent per-universe root +// sequences, the flushed multiverse roots must equal the oracle trees +// holding each universe's final root, and every returned inclusion +// proof must verify. The store and oracles persist across property +// iterations, so the comparison also covers cumulative state. +func TestMultiverseRootCoalescerProps(t *testing.T) { + t.Parallel() + + ctx := context.Background() + multiverse, _ := newTestMultiverse(t) + + oracles := map[universe.ProofType]*mssmt.CompactedTree{ + universe.ProofTypeIssuance: mssmt.NewCompactedTree( + mssmt.NewDefaultStore(), + ), + universe.ProofTypeTransfer: mssmt.NewCompactedTree( + mssmt.NewDefaultStore(), + ), + } + touched := make(map[universe.ProofType]bool) + + rapid.Check(t, func(rt *rapid.T) { + type universeUpdates struct { + id universe.Identifier + roots []mssmt.Node + } + + // Draw a set of universes with distinct identifiers: a + // universe's roots are submitted in order by a single + // goroutine, which is what entitles each submission to a + // proof of its exact leaf below. Concurrent submissions + // for the same universe may legitimately be superseded, + // so they are exercised elsewhere, not here. + numUniverses := rapid.IntRange(1, 4).Draw(rt, "num_universes") + updates := make([]universeUpdates, 0, numUniverses) + seen := make(map[universeIDKey]bool, numUniverses) + for i := 0; i < numUniverses; i++ { + var id universe.Identifier + idBytes := rapid.SliceOfN( + rapid.Byte(), 32, 32, + ).Draw(rt, "asset_id") + copy(id.AssetID[:], idBytes) + + id.ProofType = universe.ProofTypeIssuance + if rapid.Bool().Draw(rt, "is_transfer") { + id.ProofType = universe.ProofTypeTransfer + } + + if seen[id.String()] { + continue + } + seen[id.String()] = true + + numRoots := rapid.IntRange(1, 4).Draw(rt, "num_roots") + roots := make([]mssmt.Node, numRoots) + for j := range roots { + var hash mssmt.NodeHash + hashBytes := rapid.SliceOfN( + rapid.Byte(), 32, 32, + ).Draw(rt, "root_hash") + copy(hash[:], hashBytes) + + sum := rapid.Uint64Range( + 0, 1<<20, + ).Draw(rt, "root_sum") + + roots[j] = mssmt.NewComputedBranch(hash, sum) + } + + updates = append(updates, universeUpdates{ + id: id, + roots: roots, + }) + } + + // Submit all universes concurrently, each universe's + // roots in order. + type outcome struct { + update multiverseRootUpdate + err error + } + outcomes := make([][]outcome, len(updates)) + var wg sync.WaitGroup + for i := range updates { + outcomes[i] = make([]outcome, len(updates[i].roots)) + + wg.Add(1) + go func(i int) { + defer wg.Done() + + id := updates[i].id + coalescer := multiverse.rootCoalescer + for j, root := range updates[i].roots { + mvRoot, proof, err := + coalescer.updateRoot( + ctx, id, root, + ) + outcomes[i][j] = outcome{ + update: multiverseRootUpdate{ + multiverseRoot: mvRoot, + inclusionProof: proof, + }, + err: err, + } + } + }(i) + } + wg.Wait() + + // Every submission must succeed with a proof of the exact + // leaf it submitted. + for i := range updates { + id := updates[i].id + for j, root := range updates[i].roots { + out := outcomes[i][j] + require.NoError(rt, out.err) + + leaf := multiverseLeafForRoot(id, root) + require.True(rt, mssmt.VerifyMerkleProof( + id.Bytes(), leaf, + out.update.inclusionProof, + out.update.multiverseRoot, + )) + } + } + + // Feed each universe's final root to the oracle and + // compare full roots per touched proof type. + for i := range updates { + id := updates[i].id + final := updates[i].roots[len(updates[i].roots)-1] + _, err := oracles[id.ProofType].Insert( + ctx, id.Bytes(), + multiverseLeafForRoot(id, final), + ) + require.NoError(rt, err) + touched[id.ProofType] = true + } + + for proofType, oracle := range oracles { + if !touched[proofType] { + continue + } + + assertOracleRoot( + rt, ctx, multiverse, oracle, proofType, + ) + } + }) +} + +// panicFlushDB is a BatchedMultiverse whose write transactions always +// panic, simulating an unexpected failure inside a flush. +type panicFlushDB struct { + BatchedMultiverse +} + +func (panicFlushDB) ExecTx(context.Context, TxOptions, + func(BaseMultiverseStore) error) error { + + panic("flush boom") +} + +// TestMultiverseRootCoalescerFlushPanic asserts that a panic during a +// flush is contained: every waiter of the round receives an error, the +// flusher role is surrendered and no pending state leaks, so the +// coalescer keeps functioning for future updates. +func TestMultiverseRootCoalescerFlushPanic(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + var writeMu sync.Mutex + coalescer := newMultiverseRootCoalescer(panicFlushDB{}, &writeMu) + + newID := func() universe.Identifier { + var id universe.Identifier + id.ProofType = universe.ProofTypeIssuance + copy(id.AssetID[:], test.RandBytes(32)) + return id + } + var rootHash mssmt.NodeHash + copy(rootHash[:], test.RandBytes(32)) + root := mssmt.NewComputedBranch(rootHash, 1) + + // Enqueue a waiter by hand, so the panicking flush has a waiter + // beyond the leading caller itself. + waiterChan := make(chan flushResult, 1) + waiterID := newID() + coalescer.mu.Lock() + coalescer.pending[waiterID.String()] = &pendingRootUpdate{ + id: waiterID, + root: root, + waiters: []flushWaiter{{ + result: waiterChan, + wantProof: true, + }}, + } + coalescer.order = append(coalescer.order, waiterID.String()) + coalescer.mu.Unlock() + + // The leading caller must receive the panic as an error, not a + // stuck call or an unwound stack. + _, _, err := coalescer.updateRoot(ctx, newID(), root) + require.ErrorContains(t, err, "panic") + + // The bystander waiter must have been failed as well. + select { + case res := <-waiterChan: + require.ErrorContains(t, res.err, "panic") + default: + t.Fatal("waiter was not notified of the failed flush") + } + + // The flusher role and the pending queue must be clean, so + // future updates are not blocked. + coalescer.mu.Lock() + require.False(t, coalescer.flushing) + require.Empty(t, coalescer.pending) + require.Empty(t, coalescer.order) + coalescer.mu.Unlock() + + // The multiverse write lock must have been released on the way + // out, or the deletion paths would deadlock. + require.True(t, writeMu.TryLock()) + writeMu.Unlock() +} diff --git a/tapdb/multiverse_reconcile.go b/tapdb/multiverse_reconcile.go new file mode 100644 index 000000000..15b822384 --- /dev/null +++ b/tapdb/multiverse_reconcile.go @@ -0,0 +1,148 @@ +package tapdb + +import ( + "bytes" + "context" + "crypto/sha256" + "fmt" + + "github.com/lightninglabs/taproot-assets/tapdb/sqlc" + "github.com/lightninglabs/taproot-assets/universe" +) + +// ReconcileMultiverse verifies that the shared multiverse trees commit +// to the current root of every universe, and repairs any entries that +// diverged. The multiverse trees are fully derived from the universe +// roots, so a daemon that stopped between a proof insert committing +// and its multiverse update being flushed is always repairable here. +// This is intended to run at startup, before the store serves +// concurrent traffic. +func (b *MultiverseStore) ReconcileMultiverse(ctx context.Context) error { + updates, err := b.multiverseDivergence(ctx) + if err != nil { + return fmt.Errorf("unable to check multiverse "+ + "divergence: %w", err) + } + + if len(updates) == 0 { + return nil + } + + log.Warnf("Repairing %d diverged multiverse entries", len(updates)) + + err = b.rootCoalescer.updateRoots(ctx, updates) + if err != nil { + return fmt.Errorf("unable to repair multiverse entries: %w", + err) + } + + return nil +} + +// committedMultiverseLeaf is the universe root value a multiverse leaf +// currently commits to. +type committedMultiverseLeaf struct { + rootHash []byte + rootSum uint64 +} + +// multiverseDivergence returns a root update for every universe whose +// current root is not committed to by its multiverse leaf, either +// because the leaf is missing or because it holds a stale root. +func (b *MultiverseStore) multiverseDivergence( + ctx context.Context) ([]universeRootUpdate, error) { + + // Load the committed multiverse leaves for both proof types, + // keyed the same way multiverse leaf keys are derived (asset ID, + // or hash of the schnorr-serialized group key). + proofTypes := []universe.ProofType{ + universe.ProofTypeIssuance, universe.ProofTypeTransfer, + } + committed := make( + map[universe.ProofType]map[[32]byte]committedMultiverseLeaf, + len(proofTypes), + ) + + readTx := NewBaseMultiverseReadTx() + dbErr := b.db.ExecTx( + ctx, &readTx, func(db BaseMultiverseStore) error { + for _, proofType := range proofTypes { + leaves, err := db.QueryMultiverseLeaves( + ctx, QueryMultiverseLeaves{ + ProofType: proofType.String(), + }, + ) + if err != nil { + return err + } + + byKey := make( + map[[32]byte]committedMultiverseLeaf, + len(leaves), + ) + for _, leaf := range leaves { + var key [32]byte + if len(leaf.GroupKey) > 0 { + key = sha256.Sum256( + leaf.GroupKey, + ) + } else { + copy(key[:], leaf.AssetID) + } + + sum := uint64(leaf.UniverseRootSum) + byKey[key] = committedMultiverseLeaf{ + rootHash: leaf.UniverseRootHash, + rootSum: sum, + } + } + committed[proofType] = byKey + } + + return nil + }, + ) + if dbErr != nil { + return nil, dbErr + } + + // Page through all universe roots and collect those whose + // multiverse leaf doesn't commit to them. + var updates []universeRootUpdate + params := sqlc.UniverseRootsParams{ + SortDirection: sqlInt16(universe.SortAscending), + NumOffset: 0, + NumLimit: universe.RequestPageSize, + } + for { + roots, err := b.queryRootNodes(ctx, params, false) + if err != nil { + return nil, err + } + + for _, root := range roots { + expected := multiverseLeafNode(root.ID, root.Node) + expectedHash := root.Node.NodeHash() + + byKey := committed[root.ID.ProofType] + leaf, ok := byKey[root.ID.Bytes()] + if ok && bytes.Equal(leaf.rootHash, expectedHash[:]) && + leaf.rootSum == expected.NodeSum() { + + continue + } + + updates = append(updates, universeRootUpdate{ + id: root.ID, + root: root.Node, + }) + } + + params.NumOffset += universe.RequestPageSize + if len(roots) < universe.RequestPageSize { + break + } + } + + return updates, nil +} diff --git a/tapdb/multiverse_reconcile_test.go b/tapdb/multiverse_reconcile_test.go new file mode 100644 index 000000000..7638a8963 --- /dev/null +++ b/tapdb/multiverse_reconcile_test.go @@ -0,0 +1,174 @@ +package tapdb + +import ( + "context" + "testing" + + "github.com/lightninglabs/taproot-assets/internal/test" + "github.com/lightninglabs/taproot-assets/mssmt" + "github.com/lightninglabs/taproot-assets/universe" + "github.com/stretchr/testify/require" + "pgregory.net/rapid" +) + +// insertUniverseLeafOnly commits a proof leaf into its universe tree +// without updating the shared multiverse tree, simulating a daemon that +// stopped between the universe transaction committing and the +// multiverse update being written. +func insertUniverseLeafOnly(t require.TestingT, ctx context.Context, + multiverse *MultiverseStore, item *universe.Item) { + + var writeTx BaseMultiverseOptions + err := multiverse.db.ExecTx( + ctx, &writeTx, func(store BaseMultiverseStore) error { + blockHeight, err := SparseDecodeBlockHeight( + item.Leaf.RawProof, + ) + if err != nil { + return err + } + + _, err = universeUpsertProofLeaf( + ctx, store, item.ID.String(), + item.ID.ProofType, item.ID.GroupKey, item.Key, + item.Leaf, item.MetaReveal, blockHeight, + ) + + return err + }, + ) + require.NoError(t, err) +} + +// tamperMultiverseLeaf overwrites a universe's multiverse leaf with a +// bogus root, diverging it from the universe's actual root. +func tamperMultiverseLeaf(t require.TestingT, ctx context.Context, + multiverse *MultiverseStore, id universe.Identifier) { + + var bogusHash mssmt.NodeHash + copy(bogusHash[:], test.RandBytes(32)) + bogusRoot := mssmt.NewComputedBranch( + bogusHash, uint64(test.RandInt[uint32]()), + ) + + err := multiverse.rootCoalescer.updateRoots( + ctx, []universeRootUpdate{{id: id, root: bogusRoot}}, + ) + require.NoError(t, err) +} + +// TestReconcileMultiverse asserts that startup reconciliation detects +// and repairs both kinds of divergence between the universe trees and +// the shared multiverse trees: a missing multiverse leaf (crash between +// universe commit and multiverse write) and a stale one. +func TestReconcileMultiverse(t *testing.T) { + t.Parallel() + + ctx := context.Background() + multiverse, _ := newTestMultiverse(t) + + // A healthy store must show no divergence, including when empty. + updates, err := multiverse.multiverseDivergence(ctx) + require.NoError(t, err) + require.Empty(t, updates) + + // Insert a few items through the normal path. + items := make([]*universe.Item, 3) + for i := range items { + items[i] = genRandomAsset(t) + } + require.NoError(t, multiverse.UpsertProofLeafBatch(ctx, items)) + + updates, err = multiverse.multiverseDivergence(ctx) + require.NoError(t, err) + require.Empty(t, updates) + + // Orphan a new universe: leaf committed, multiverse never + // updated. + orphan := genRandomAsset(t) + insertUniverseLeafOnly(t, ctx, multiverse, orphan) + + // Tamper an existing universe's multiverse entry. + tamperMultiverseLeaf(t, ctx, multiverse, items[0].ID) + + updates, err = multiverse.multiverseDivergence(ctx) + require.NoError(t, err) + require.Len(t, updates, 2) + + // Reconciliation must repair both. + require.NoError(t, multiverse.ReconcileMultiverse(ctx)) + + updates, err = multiverse.multiverseDivergence(ctx) + require.NoError(t, err) + require.Empty(t, updates) + + // Reconciling a healthy store must be a no-op. + require.NoError(t, multiverse.ReconcileMultiverse(ctx)) +} + +// TestReconcileMultiverseProps property-tests reconciliation: from any +// mix of healthy universes, orphaned universes (universe leaf committed +// but multiverse never updated) and tampered multiverse entries, +// ReconcileMultiverse must restore a state with no divergence. The +// store persists across property iterations, so repairs are also +// checked against cumulative state. +func TestReconcileMultiverseProps(t *testing.T) { + t.Parallel() + + ctx := context.Background() + multiverse, _ := newTestMultiverse(t) + + rapid.Check(t, func(rt *rapid.T) { + // Draw a batch of new universes, each healthy, orphaned or + // with a tampered multiverse entry. + numUniverses := rapid.IntRange(1, 4).Draw(rt, "num_universes") + + var expectDiverged int + for i := 0; i < numUniverses; i++ { + item := genRandomAsset(t) + + kind := rapid.SampledFrom([]string{ + "healthy", "orphaned", "tampered", + }).Draw(rt, "kind") + + switch kind { + case "healthy": + _, err := multiverse.UpsertProofLeaf( + ctx, item.ID, item.Key, item.Leaf, + item.MetaReveal, + ) + require.NoError(rt, err) + + case "orphaned": + insertUniverseLeafOnly( + rt, ctx, multiverse, item, + ) + expectDiverged++ + + case "tampered": + _, err := multiverse.UpsertProofLeaf( + ctx, item.ID, item.Key, item.Leaf, + item.MetaReveal, + ) + require.NoError(rt, err) + + tamperMultiverseLeaf( + rt, ctx, multiverse, item.ID, + ) + expectDiverged++ + } + } + + // Exactly the orphaned and tampered universes must show + // up as diverged: the previous iteration ended reconciled. + updates, err := multiverse.multiverseDivergence(ctx) + require.NoError(rt, err) + require.Len(rt, updates, expectDiverged) + + require.NoError(rt, multiverse.ReconcileMultiverse(ctx)) + + updates, err = multiverse.multiverseDivergence(ctx) + require.NoError(rt, err) + require.Empty(rt, updates) + }) +} diff --git a/tapdb/universe.go b/tapdb/universe.go index 932b27df9..a12d3f00d 100644 --- a/tapdb/universe.go +++ b/tapdb/universe.go @@ -773,13 +773,21 @@ func (b *BaseUniverseTree) UpsertProofLeaf(ctx context.Context, return fmt.Errorf("failed universe upsert: %w", err) } - multiRoot, multiProof, err := upsertMultiverseLeafEntry( + err = upsertMultiverseLeafEntry( ctx, dbTx, b.id, issuanceProof.UniverseRoot, ) if err != nil { return fmt.Errorf("failed multiverse upsert: %w", err) } + multiRoot, multiProof, err := multiverseRootAndProof( + ctx, dbTx, b.id, + ) + if err != nil { + return fmt.Errorf("failed multiverse root fetch: %w", + err) + } + issuanceProof.MultiverseRoot = multiRoot issuanceProof.MultiverseInclusionProof = multiProof @@ -793,20 +801,37 @@ func (b *BaseUniverseTree) UpsertProofLeaf(ctx context.Context, return uniProof, nil } +// multiverseLeafNode builds the multiverse leaf committing to the given +// universe root. For issuance proofs, the leaf sum is always 1 (one asset or +// group). For transfers, it's the universe root's actual amount. +func multiverseLeafNode(id universe.Identifier, + universeRoot mssmt.Node) *mssmt.LeafNode { + + universeRootHash := universeRoot.NodeHash() + assetGroupSum := universeRoot.NodeSum() + + if id.ProofType == universe.ProofTypeIssuance { + assetGroupSum = 1 + } + + return mssmt.NewLeafNode(universeRootHash[:], assetGroupSum) +} + // upsertMultiverseLeafEntry inserts the universe root into the main multiverse // tree. This should be called *after* universeUpsertProofLeaf if the proof -// needs to be added to the main issuance/transfer multiverse. +// needs to be added to the main issuance/transfer multiverse. The resulting +// multiverse root and inclusion proof can be fetched separately with +// multiverseRootAndProof. // // NOTE: This function accepts a db transaction, as it's used when making // broader DB updates. func upsertMultiverseLeafEntry(ctx context.Context, dbTx BaseUniverseStore, - id universe.Identifier, universeRoot mssmt.Node) ( - mssmt.Node, *mssmt.Proof, error) { + id universe.Identifier, universeRoot mssmt.Node) error { // Determine the multiverse namespace based on the proof type. multiverseNS, err := namespaceForProof(id.ProofType) if err != nil { - return nil, nil, err + return err } // Retrieve a handle to the multiverse tree. @@ -815,24 +840,14 @@ func upsertMultiverseLeafEntry(ctx context.Context, dbTx BaseUniverseStore, ) // Construct a leaf node for insertion into the multiverse tree. - universeRootHash := universeRoot.NodeHash() - assetGroupSum := universeRoot.NodeSum() - - // For issuance proofs, the sum in the multiverse is always 1 (one asset - // or group). For transfers, it's the actual amount. - if id.ProofType == universe.ProofTypeIssuance { - assetGroupSum = 1 - } - - uniLeafNode := mssmt.NewLeafNode(universeRootHash[:], assetGroupSum) + uniLeafNode := multiverseLeafNode(id, universeRoot) // Use asset ID (or asset group hash) as the upper tree leaf node key. uniLeafNodeKey := id.Bytes() _, err = multiverseTree.Insert(ctx, uniLeafNodeKey, uniLeafNode) if err != nil { - return nil, nil, fmt.Errorf("multiverse tree insert "+ - "failed: %w", err) + return fmt.Errorf("multiverse tree insert failed: %w", err) } // Ensure the corresponding multiverse roots and leaves DB entries @@ -844,8 +859,7 @@ func upsertMultiverseLeafEntry(ctx context.Context, dbTx BaseUniverseStore, }, ) if err != nil { - return nil, nil, fmt.Errorf("unable to upsert multiverse "+ - "root: %w", err) + return fmt.Errorf("unable to upsert multiverse root: %w", err) } var assetIDBytes, groupKeyBytes []byte @@ -863,26 +877,47 @@ func upsertMultiverseLeafEntry(ctx context.Context, dbTx BaseUniverseStore, LeafNodeNamespace: multiverseNS, }) if err != nil { - return nil, nil, fmt.Errorf("unable to upsert multiverse "+ - "leaf: %w", err) + return fmt.Errorf("unable to upsert multiverse leaf: %w", err) + } + + return nil +} + +// multiverseRootAndProof returns the current root of the multiverse tree for +// the given universe's proof type, along with the inclusion proof for the +// universe's leaf within it. +// +// NOTE: This function accepts a db transaction, as it's used when making +// broader DB updates. +func multiverseRootAndProof(ctx context.Context, dbTx BaseUniverseStore, + id universe.Identifier) (mssmt.Node, *mssmt.Proof, error) { + + // Determine the multiverse namespace based on the proof type. + multiverseNS, err := namespaceForProof(id.ProofType) + if err != nil { + return nil, nil, err } - // Retrieve the multiverse root and inclusion proof. - finalMultiverseRoot, err := multiverseTree.Root(ctx) + // Retrieve a handle to the multiverse tree. + multiverseTree := mssmt.NewCompactedTree( + newTreeStoreWrapperTx(dbTx, multiverseNS), + ) + + multiverseRoot, err := multiverseTree.Root(ctx) if err != nil { return nil, nil, fmt.Errorf("failed to get multiverse "+ "root: %w", err) } multiverseInclusionProof, err := multiverseTree.MerkleProof( - ctx, uniLeafNodeKey, + ctx, id.Bytes(), ) if err != nil { return nil, nil, fmt.Errorf("failed to get multiverse "+ "proof: %w", err) } - return finalMultiverseRoot, multiverseInclusionProof, nil + return multiverseRoot, multiverseInclusionProof, nil } // universeUpsertProofLeaf upserts a proof leaf within the universe tree (stored diff --git a/tapdb/universe_test.go b/tapdb/universe_test.go index 7e36af68c..4bad992de 100644 --- a/tapdb/universe_test.go +++ b/tapdb/universe_test.go @@ -1893,3 +1893,84 @@ func TestProofCacheInvalidatesOnSiblingInsert(t *testing.T) { "current universe root", ) } + +// TestUpsertProofLeafBatchMultiverseRoot asserts that the batch insert +// path produces the same multiverse roots as inserting the same items +// one at a time, including when a batch contains multiple items for the +// same universe (in which case only the universe's final root must be +// reflected in the multiverse tree). +func TestUpsertProofLeafBatchMultiverseRoot(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + batchStore, _ := newTestMultiverse(t) + serialStore, _ := newTestMultiverse(t) + + // Generate random items, then add a second leaf to each universe + // so the batch contains multiple items per universe. + const numAssets = 5 + var items []*universe.Item + for i := 0; i < numAssets; i++ { + item := genRandomAsset(t) + items = append(items, item) + + leaf := randMintingLeaf( + t, item.Leaf.Genesis, item.ID.GroupKey, + ) + if item.ID.ProofType == universe.ProofTypeTransfer { + prevWitnesses := leaf.Asset.PrevWitnesses + prevWitnesses[0].TxWitness = [][]byte{ + {1}, {1}, {1}, + } + prevID := prevWitnesses[0].PrevID + prevID.OutPoint.Hash = [32]byte{1} + } + + items = append(items, &universe.Item{ + ID: item.ID, + Key: randLeafKey(t), + Leaf: &leaf, + }) + } + + err := batchStore.UpsertProofLeafBatch(ctx, items) + require.NoError(t, err) + + for _, item := range items { + _, err := serialStore.UpsertProofLeaf( + ctx, item.ID, item.Key, item.Leaf, item.MetaReveal, + ) + require.NoError(t, err) + } + + proofTypes := []universe.ProofType{ + universe.ProofTypeIssuance, universe.ProofTypeTransfer, + } + for _, proofType := range proofTypes { + batchRoot, err := batchStore.MultiverseRootNode( + ctx, proofType, + ) + require.NoError(t, err) + serialRoot, err := serialStore.MultiverseRootNode( + ctx, proofType, + ) + require.NoError(t, err) + + require.Equal( + t, serialRoot.IsSome(), batchRoot.IsSome(), + "proof type %v", proofType, + ) + + serialRoot.WhenSome(func(sr universe.MultiverseRoot) { + batchRoot.WhenSome(func(br universe.MultiverseRoot) { + require.True( + t, mssmt.IsEqualNode(sr.Node, br.Node), + "proof type %v: serial root %v != "+ + "batch root %v", proofType, + sr.Node, br.Node, + ) + }) + }) + } +}