From feed30849da375da8a39d333a5242e049a7279c5 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 17:48:06 -0230 Subject: [PATCH 1/8] mssmt: add property test for last-write-wins insert semantics Inserting a sequence of leaves that reuses keys must produce the same tree as inserting only the final leaf observed per key. This is the invariant that permits coalescing consecutive updates to the same key into a single insert of the latest value, as the multiverse tree does for universe roots. --- mssmt/tree_prop_test.go | 99 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 mssmt/tree_prop_test.go 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) +} From 3784cda0900cb1eb4ca65a18cd4b8d7194694347 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 17:55:55 -0230 Subject: [PATCH 2/8] tapdb: update multiverse tree once per universe in batch inserts Split upsertMultiverseLeafEntry into its write half and a separate multiverseRootAndProof read helper. The batch insert path previously called the combined function once per item, computing a multiverse root and inclusion proof each time only to discard them (the batch callers never read those fields), and rewriting the same universe's multiverse leaf once per item. UpsertProofLeafBatch now tracks the final universe root per universe and upserts each universe's multiverse leaf exactly once. This is sound because SMT insertion is last-write-wins per key (see the mssmt property test). DeleteProofLeaf similarly stops computing a discarded root and proof. Single-leaf paths keep their semantics: they call the write half and then fetch the root and inclusion proof explicitly. --- tapdb/multiverse.go | 60 +++++++++++++++++++++---------- tapdb/universe.go | 58 ++++++++++++++++++++++-------- tapdb/universe_test.go | 81 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 166 insertions(+), 33 deletions(-) diff --git a/tapdb/multiverse.go b/tapdb/multiverse.go index 0176c37e9..919aa4491 100644 --- a/tapdb/multiverse.go +++ b/tapdb/multiverse.go @@ -815,15 +815,21 @@ func (b *MultiverseStore) UpsertProofLeaf(ctx context.Context, // Now, attempt to insert the universe root into the main // multiverse tree. - // - // nolint:lll - multiverseRoot, multiverseProof, err = upsertMultiverseLeafEntry( + err = upsertMultiverseLeafEntry( ctx, dbTx, id, uniProof.UniverseRoot, ) if err != nil { return fmt.Errorf("failed multiverse upsert: %w", err) } + multiverseRoot, multiverseProof, err = multiverseRootAndProof( + ctx, dbTx, id, + ) + if err != nil { + return fmt.Errorf("failed multiverse root fetch: %w", + err) + } + return nil } dbErr := b.db.ExecTx(ctx, &writeTx, execTxFunc) @@ -872,9 +878,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. + type multiverseUpdate struct { + id universe.Identifier + root mssmt.Node + } + dbErr := b.db.ExecTx( ctx, &writeTx, func(store BaseMultiverseStore) error { uniProofs = make([]*universe.Proof, len(items)) + + finalRoots := make( + map[universeIDKey]*multiverseUpdate, + len(items), + ) + var updateOrder []universeIDKey + for idx := range items { item := items[idx] @@ -902,24 +923,27 @@ 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, + key := item.ID.String() + if _, ok := finalRoots[key]; !ok { + updateOrder = append(updateOrder, key) + } + finalRoots[key] = &multiverseUpdate{ + id: item.ID, + root: uniProof.UniverseRoot, + } + } + + // Next, we'll insert each universe's final root into + // the main multiverse tree. + for _, key := range updateOrder { + update := finalRoots[key] + err := upsertMultiverseLeafEntry( + ctx, store, update.id, update.root, ) if err != nil { return fmt.Errorf("failed multiverse "+ - "upsert for item %d: %w", - idx, err) + "upsert for %v: %w", key, err) } - - // Update the proof object with multiverse - // details. - uniProofs[idx].MultiverseRoot = multiRoot - uniProofs[idx].MultiverseInclusionProof = multiProof //nolint:lll } return nil @@ -1060,7 +1084,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/universe.go b/tapdb/universe.go index 932b27df9..28691ea2b 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 @@ -795,18 +803,19 @@ func (b *BaseUniverseTree) UpsertProofLeaf(ctx context.Context, // 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. @@ -831,8 +840,7 @@ func upsertMultiverseLeafEntry(ctx context.Context, dbTx BaseUniverseStore, _, 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 +852,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 +870,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) } - // Retrieve the multiverse root and inclusion proof. - finalMultiverseRoot, err := multiverseTree.Root(ctx) + 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 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, + ) + }) + }) + } +} From 7b309c4edb25d0d9bb19a8e33b282cb2094ce8fc Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 18:02:06 -0230 Subject: [PATCH 3/8] tapdb: add multiverse root write coalescer 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, regardless of which universes they touch: under Postgres serializable isolation one of them aborts and retries with backoff, effectively serializing ingest across universes (and making it slower than actually serial, since the retry backoff starts at 20-60ms). The coalescer funnels all multiverse writes through a single flusher using leader-based group commit: the first caller to find it idle flushes pending updates in rounds until none remain, while other callers just await their result. Updates accumulate while a flush is in flight and are applied together in one transaction, at most one leaf write per universe (last-write-wins per SMT key). Waiters receive the post-flush multiverse root and their universe's inclusion proof, preserving the response semantics of the insert paths. Not yet wired into the insert paths; this commit adds the component, its concurrency unit test, and a property test comparing flushed state against an in-memory oracle tree. --- tapdb/multiverse.go | 6 + tapdb/multiverse_coalescer.go | 258 ++++++++++++++++++++ tapdb/multiverse_coalescer_test.go | 368 +++++++++++++++++++++++++++++ 3 files changed, 632 insertions(+) create mode 100644 tapdb/multiverse_coalescer.go create mode 100644 tapdb/multiverse_coalescer_test.go diff --git a/tapdb/multiverse.go b/tapdb/multiverse.go index 919aa4491..7b6b75343 100644 --- a/tapdb/multiverse.go +++ b/tapdb/multiverse.go @@ -132,6 +132,11 @@ 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 + // 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 @@ -164,6 +169,7 @@ func NewMultiverseStore(db BatchedMultiverse, cfg.Caches.LeavesNumCachedUniverses, cfg.Caches.LeavesPerUniverse, ), + rootCoalescer: newMultiverseRootCoalescer(db), transferProofDistributor: fn.NewEventDistributor[proof.Blob](), }, nil } diff --git a/tapdb/multiverse_coalescer.go b/tapdb/multiverse_coalescer.go new file mode 100644 index 000000000..aa80e3fe0 --- /dev/null +++ b/tapdb/multiverse_coalescer.go @@ -0,0 +1,258 @@ +package tapdb + +import ( + "context" + "fmt" + "runtime/debug" + "sync" + + "github.com/lightninglabs/taproot-assets/mssmt" + "github.com/lightninglabs/taproot-assets/universe" +) + +// 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 +} + +// 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 []chan flushResult +} + +// 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 + + 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. +func newMultiverseRootCoalescer( + db BatchedMultiverse) *multiverseRootCoalescer { + + return &multiverseRootCoalescer{ + db: db, + 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) { + + result := make(chan flushResult, 1) + + c.mu.Lock() + 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, result) + + 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() + } +} + +// 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 <- 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() + + var writeTx BaseMultiverseOptions + results := make([]multiverseRootUpdate, len(batch)) + err := c.db.ExecTx( + ctx, &writeTx, func(store BaseMultiverseStore) error { + // Write every universe's latest root first, then + // read back the resulting root and one inclusion + // proof per universe. + 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 { + 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 <- res + } + } +} diff --git a/tapdb/multiverse_coalescer_test.go b/tapdb/multiverse_coalescer_test.go new file mode 100644 index 000000000..55cc211ef --- /dev/null +++ b/tapdb/multiverse_coalescer_test.go @@ -0,0 +1,368 @@ +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) + } +} + +// 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() + coalescer := newMultiverseRootCoalescer(panicFlushDB{}) + + 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: []chan flushResult{waiterChan}, + } + 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() +} From d34df704b9a461a32868f3c67a0f51d6fad606d5 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 18:06:37 -0230 Subject: [PATCH 4/8] tapdb: route single-leaf multiverse updates through the coalescer MultiverseStore.UpsertProofLeaf now commits only the per-universe rows in its own transaction, then reflects the universe's new root in the shared multiverse tree via the root coalescer. Insert transactions for different universes no longer touch any shared rows, so they can commit in parallel on Postgres instead of aborting each other through serialization failures on the multiverse root. Response semantics are preserved: the caller still receives the multiverse root and inclusion proof, now from the flush that carried its update. Under concurrent inserts into the same universe, those fields may reflect a slightly newer universe root than the one in the same response, which is the accurate post-flush state. A failed flush leaves the universe leaf committed and the multiverse entry stale; the entry is healed by the universe's next successful update. BaseUniverseTree.UpsertProofLeaf keeps its inline multiverse write, as it has no production callers (tests and bench fixtures only). --- tapdb/multiverse.go | 39 +++++++++++++++++---------------------- 1 file changed, 17 insertions(+), 22 deletions(-) diff --git a/tapdb/multiverse.go b/tapdb/multiverse.go index 7b6b75343..7ec2502a8 100644 --- a/tapdb/multiverse.go +++ b/tapdb/multiverse.go @@ -796,10 +796,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 { @@ -819,23 +817,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. - err = upsertMultiverseLeafEntry( - ctx, dbTx, id, uniProof.UniverseRoot, - ) - if err != nil { - return fmt.Errorf("failed multiverse upsert: %w", err) - } - - multiverseRoot, multiverseProof, err = multiverseRootAndProof( - ctx, dbTx, id, - ) - if err != nil { - return fmt.Errorf("failed multiverse root fetch: %w", - err) - } - return nil } dbErr := b.db.ExecTx(ctx, &writeTx, execTxFunc) @@ -843,8 +824,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 From f7c18e1ca121df7444525398ad444d722beea115 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 19:27:43 -0230 Subject: [PATCH 5/8] tapdb: route batch multiverse updates through the coalescer UpsertProofLeafBatch now commits only the per-universe rows in its own transaction, then submits each universe's final root to the root coalescer, matching the single-leaf path. With this, every insert-path write to the shared multiverse trees flows through the coalescer, so insert transactions never contend on rows shared across universes. Batch callers never consume multiverse roots or inclusion proofs, so the new batch entry point registers waiters that skip generating them; a flush only computes the root and proof for universes with a single-leaf waiter attached. Deletion paths (DeleteProofLeaf, DeleteUniverse) keep their inline multiverse writes: they are rare administrative operations, and any collision with a flush is absorbed by the existing transaction retry. --- tapdb/multiverse.go | 45 ++++++----- tapdb/multiverse_coalescer.go | 118 +++++++++++++++++++++++++---- tapdb/multiverse_coalescer_test.go | 109 +++++++++++++++++++++++++- 3 files changed, 232 insertions(+), 40 deletions(-) diff --git a/tapdb/multiverse.go b/tapdb/multiverse.go index 7ec2502a8..d5bfa3854 100644 --- a/tapdb/multiverse.go +++ b/tapdb/multiverse.go @@ -882,20 +882,20 @@ func (b *MultiverseStore) UpsertProofLeafBatch(ctx context.Context, // 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. - type multiverseUpdate struct { - id universe.Identifier - root mssmt.Node - } + 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]*multiverseUpdate, + finalRoots = make( + map[universeIDKey]universeRootUpdate, len(items), ) - var updateOrder []universeIDKey + updateOrder = nil for idx := range items { item := items[idx] @@ -928,25 +928,12 @@ func (b *MultiverseStore) UpsertProofLeafBatch(ctx context.Context, if _, ok := finalRoots[key]; !ok { updateOrder = append(updateOrder, key) } - finalRoots[key] = &multiverseUpdate{ + finalRoots[key] = universeRootUpdate{ id: item.ID, root: uniProof.UniverseRoot, } } - // Next, we'll insert each universe's final root into - // the main multiverse tree. - for _, key := range updateOrder { - update := finalRoots[key] - err := upsertMultiverseLeafEntry( - ctx, store, update.id, update.root, - ) - if err != nil { - return fmt.Errorf("failed multiverse "+ - "upsert for %v: %w", key, err) - } - } - return nil }, ) @@ -954,6 +941,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() diff --git a/tapdb/multiverse_coalescer.go b/tapdb/multiverse_coalescer.go index aa80e3fe0..994446346 100644 --- a/tapdb/multiverse_coalescer.go +++ b/tapdb/multiverse_coalescer.go @@ -31,6 +31,21 @@ type flushResult struct { 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 @@ -40,7 +55,7 @@ type flushResult struct { type pendingRootUpdate struct { id universe.Identifier root mssmt.Node - waiters []chan flushResult + waiters []flushWaiter } // multiverseRootCoalescer serializes all writes to the shared @@ -106,19 +121,8 @@ func (c *multiverseRootCoalescer) updateRoot(ctx context.Context, id universe.Identifier, root mssmt.Node) (mssmt.Node, *mssmt.Proof, error) { - result := make(chan flushResult, 1) - c.mu.Lock() - 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, result) - + result := c.enqueue(id, root, true) lead := !c.flushing if lead { c.flushing = true @@ -143,6 +147,73 @@ func (c *multiverseRootCoalescer) updateRoot(ctx context.Context, } } +// 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() { @@ -192,7 +263,7 @@ func (c *multiverseRootCoalescer) flushBatch(batch []*pendingRootUpdate) { // normal delivery path below. for _, waiter := range update.waiters { select { - case waiter <- res: + case waiter.result <- res: default: } } @@ -204,13 +275,24 @@ func (c *multiverseRootCoalescer) flushBatch(batch []*pendingRootUpdate) { // 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 + } + var writeTx BaseMultiverseOptions results := make([]multiverseRootUpdate, len(batch)) err := c.db.ExecTx( ctx, &writeTx, func(store BaseMultiverseStore) error { // Write every universe's latest root first, then // read back the resulting root and one inclusion - // proof per universe. + // proof per universe that has a waiter consuming + // them. for _, update := range batch { err := upsertMultiverseLeafEntry( ctx, store, update.id, update.root, @@ -223,6 +305,10 @@ func (c *multiverseRootCoalescer) flushBatch(batch []*pendingRootUpdate) { } for i, update := range batch { + if !wantProof(update) { + continue + } + root, proof, err := multiverseRootAndProof( ctx, store, update.id, ) @@ -252,7 +338,7 @@ func (c *multiverseRootCoalescer) flushBatch(batch []*pendingRootUpdate) { // blocks the flusher, even if a waiter has abandoned its // call due to context cancellation. for _, waiter := range update.waiters { - waiter <- res + waiter.result <- res } } } diff --git a/tapdb/multiverse_coalescer_test.go b/tapdb/multiverse_coalescer_test.go index 55cc211ef..51d909aba 100644 --- a/tapdb/multiverse_coalescer_test.go +++ b/tapdb/multiverse_coalescer_test.go @@ -148,6 +148,106 @@ func TestMultiverseRootCoalescer(t *testing.T) { } } +// 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 @@ -338,9 +438,12 @@ func TestMultiverseRootCoalescerFlushPanic(t *testing.T) { waiterID := newID() coalescer.mu.Lock() coalescer.pending[waiterID.String()] = &pendingRootUpdate{ - id: waiterID, - root: root, - waiters: []chan flushResult{waiterChan}, + id: waiterID, + root: root, + waiters: []flushWaiter{{ + result: waiterChan, + wantProof: true, + }}, } coalescer.order = append(coalescer.order, waiterID.String()) coalescer.mu.Unlock() From 2b6b502d14bee044310bff0b3202d6d9583a485f Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 19:36:29 -0230 Subject: [PATCH 6/8] tapdb: reconcile multiverse from universe roots at startup Since multiverse updates are written outside the proof insert transaction, a daemon stopping between a universe commit and its multiverse flush leaves the shared tree committing to a stale root, or missing the universe's leaf entirely. The multiverse trees are fully derived data (each leaf commits to a universe root), so this is always repairable. ReconcileMultiverse compares every universe root against its multiverse leaf and rewrites diverged entries through the root coalescer. It runs during server construction, before the store serves concurrent traffic. The leaf construction rule is extracted into multiverseLeafNode so the insert path and the reconciliation check share it. Covered by a deterministic crash-window test and a property test mixing healthy, orphaned and tampered universes. --- tapcfg/server.go | 9 ++ tapdb/multiverse_reconcile.go | 148 ++++++++++++++++++++++++ tapdb/multiverse_reconcile_test.go | 174 +++++++++++++++++++++++++++++ tapdb/universe.go | 27 +++-- 4 files changed, 348 insertions(+), 10 deletions(-) create mode 100644 tapdb/multiverse_reconcile.go create mode 100644 tapdb/multiverse_reconcile_test.go 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/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 28691ea2b..a12d3f00d 100644 --- a/tapdb/universe.go +++ b/tapdb/universe.go @@ -801,6 +801,22 @@ 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. The resulting @@ -824,16 +840,7 @@ 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() From 08257d0e1b09e555124c77bdd4286a105aa51ead Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 19:58:26 -0230 Subject: [PATCH 7/8] tapdb: run multiverse flushes at read committed on Postgres With the multiverse root row moved out of the insert transactions, the remaining concurrency bottleneck is the flush transaction itself: under serializable isolation its SMT walks take page-level predicate locks (the recursive CTE plans as a bitmap index scan) that false-share index pages with every in-flight universe transaction, so flushes abort and retry with backoff, stalling all of their waiters. The flush is the sole writer of the multiverse namespaces, so serializable isolation buys it nothing: run it at read committed on Postgres. A non-serializable writer takes no predicate locks and does not flag conflicts on serializable readers, so flushes can neither abort nor be aborted by concurrent universe transactions. The single-writer invariant that makes this safe is enforced by the coalescer's single-flusher role plus a process-wide multiverse write mutex now shared with the deletion paths, the only other multiverse writers. Benchmarked on Postgres 15 (docker, 1k pre-populated universes, 8 workers inserting into distinct universes): concurrent ingest goes from 21.8 leaves/s at the merge base (worse than its 57.0 leaves/s serial, due to serialization-failure backoff) to 65.7 leaves/s, a 3x improvement, with no remaining inversion. SQLite is unaffected: the isolation override is Postgres-only. --- tapdb/interfaces.go | 19 ++++++++++++- tapdb/multiverse.go | 28 ++++++++++++++++--- tapdb/multiverse_coalescer.go | 43 ++++++++++++++++++++++++++---- tapdb/multiverse_coalescer_test.go | 9 ++++++- 4 files changed, 89 insertions(+), 10 deletions(-) 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 d5bfa3854..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" @@ -137,6 +138,12 @@ type MultiverseStore struct { // 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 @@ -154,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( @@ -169,9 +176,13 @@ func NewMultiverseStore(db BatchedMultiverse, cfg.Caches.LeavesNumCachedUniverses, cfg.Caches.LeavesPerUniverse, ), - rootCoalescer: newMultiverseRootCoalescer(db), 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 @@ -1006,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 { @@ -1024,6 +1039,7 @@ func (b *MultiverseStore) DeleteUniverse(ctx context.Context, return deleteUniverseTree(ctx, tx, id) }) + b.multiverseWriteMu.Unlock() if dbErr != nil { return "", dbErr } @@ -1047,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() diff --git a/tapdb/multiverse_coalescer.go b/tapdb/multiverse_coalescer.go index 994446346..ba44354d5 100644 --- a/tapdb/multiverse_coalescer.go +++ b/tapdb/multiverse_coalescer.go @@ -2,6 +2,7 @@ package tapdb import ( "context" + "database/sql" "fmt" "runtime/debug" "sync" @@ -10,6 +11,26 @@ import ( "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 @@ -82,6 +103,13 @@ type pendingRootUpdate struct { 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. @@ -98,12 +126,14 @@ type multiverseRootCoalescer struct { } // newMultiverseRootCoalescer creates a new coalescer that writes -// through the given db handle. -func newMultiverseRootCoalescer( - db BatchedMultiverse) *multiverseRootCoalescer { +// 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), } } @@ -285,10 +315,13 @@ func (c *multiverseRootCoalescer) flushBatch(batch []*pendingRootUpdate) { return false } - var writeTx BaseMultiverseOptions + c.writeMu.Lock() + defer c.writeMu.Unlock() + results := make([]multiverseRootUpdate, len(batch)) + flushTx := multiverseFlushTx{} err := c.db.ExecTx( - ctx, &writeTx, func(store BaseMultiverseStore) error { + 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 diff --git a/tapdb/multiverse_coalescer_test.go b/tapdb/multiverse_coalescer_test.go index 51d909aba..6bea106f9 100644 --- a/tapdb/multiverse_coalescer_test.go +++ b/tapdb/multiverse_coalescer_test.go @@ -420,7 +420,9 @@ func TestMultiverseRootCoalescerFlushPanic(t *testing.T) { t.Parallel() ctx := context.Background() - coalescer := newMultiverseRootCoalescer(panicFlushDB{}) + + var writeMu sync.Mutex + coalescer := newMultiverseRootCoalescer(panicFlushDB{}, &writeMu) newID := func() universe.Identifier { var id universe.Identifier @@ -468,4 +470,9 @@ func TestMultiverseRootCoalescerFlushPanic(t *testing.T) { 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() } From 0ee23fda112eb19434a779331cc7a08ba76942e2 Mon Sep 17 00:00:00 2001 From: Jared Tobin Date: Sat, 4 Jul 2026 20:25:10 -0230 Subject: [PATCH 8/8] docs: add release note --- docs/release-notes/release-notes-0.8.2.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) 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