diff --git a/decentralized-api/internal/server/public/poc_handler.go b/decentralized-api/internal/server/public/poc_handler.go index bfe1bf8ee..9fb62a083 100644 --- a/decentralized-api/internal/server/public/poc_handler.go +++ b/decentralized-api/internal/server/public/poc_handler.go @@ -406,34 +406,11 @@ func (s *Server) preparePocProofRequest( reqCount := uint32(req.Count) - // Quota on distinct snapshot counts per validator (after signature - // verification, so only authenticated requests consume quota). Honest - // validation touches at most the early and final snapshots; cycling - // through more counts is a rebuild-DoS probe, not a legitimate pattern. - if s.pocSnapshotLimiter != nil { - allowed, distinct := s.pocSnapshotLimiter.Allow(req.ValidatorAddress, int64(req.PocStageStartBlockHeight), req.ModelId, reqCount) - if !allowed { - logging.Warn("PoC proofs snapshot-count quota exceeded", types.Validation, - "validatorAddress", req.ValidatorAddress, - "pocStageStartBlockHeight", req.PocStageStartBlockHeight, - "modelId", req.ModelId, - "requestedCount", reqCount, - "distinctCounts", distinct) - return nil, 0, nil, echo.NewHTTPError(http.StatusTooManyRequests, "too many distinct snapshot counts requested") - } - // Two distinct counts (early + final) is the expected honest ceiling. - // A third is tolerated but anomalous: surface it so operators can - // spot validators whose early capture diverged (or probing) before - // the quota ever trips. - if distinct > 2 { - logging.Warn("PoC proofs validator exceeded expected two snapshot counts", types.Validation, - "validatorAddress", req.ValidatorAddress, - "pocStageStartBlockHeight", req.PocStageStartBlockHeight, - "modelId", req.ModelId, - "requestedCount", reqCount, - "distinctCounts", distinct) - } - } + // No per-validator snapshot-count quota is needed: committed counts are + // served from retained copy-on-write snapshots in O(depth), and any + // non-committed count is rejected outright (its root can never match the + // on-chain commitment), so a proof request can no longer trigger an O(N) + // rebuild — the rebuild-DoS surface the quota guarded against is gone. storeRoot, err := stageStore.GetRootAt(reqCount) if err != nil { logging.Warn("Snapshot count not servable", types.Validation, diff --git a/decentralized-api/internal/server/public/poc_handler_test.go b/decentralized-api/internal/server/public/poc_handler_test.go index 5e7276306..4eefc717f 100644 --- a/decentralized-api/internal/server/public/poc_handler_test.go +++ b/decentralized-api/internal/server/public/poc_handler_test.go @@ -494,87 +494,3 @@ func TestPostPocProofsByNonce_UsesModelScopedStore(t *testing.T) { assert.Equal(t, uint32(0), resp.Proofs[0].LeafIndex) mockRecorder.AssertExpectations(t) } - -func TestPostPocProofs_DistinctSnapshotCountQuota(t *testing.T) { - store := artifacts.NewManagedArtifactStore(t.TempDir(), 3) - defer store.Close() - - // Four flush boundaries: counts 1..4, each a legitimately servable snapshot. - modelStore, err := store.GetOrCreateStore(100, "model-a") - assert.NoError(t, err) - for i := int32(1); i <= 4; i++ { - assert.NoError(t, modelStore.AddWithNode(i, []byte{byte(i), 2, 3, 4}, "")) - assert.NoError(t, modelStore.Flush()) - } - - privKey := secp256k1.GenPrivKey() - pubKeyB64 := base64.StdEncoding.EncodeToString(privKey.PubKey().Bytes()) - queryClient, cleanup := newProofQueryClient(t, &proofQueryServer{pubkeyB64: pubKeyB64}) - defer cleanup() - - mockRecorder := &cosmosclient.MockCosmosMessageClient{} - mockRecorder.On("NewInferenceQueryClient").Return(queryClient) - mockRecorder.On("GetSignerAddress").Return("participant-signer") - mockRecorder.On("SignBytes", mock.Anything).Return([]byte("response-signature"), nil) - - server := &Server{ - artifactStore: store, - recorder: mockRecorder, - authzCache: authzcache.NewAuthzCache(mockRecorder), - pocSnapshotLimiter: newSnapshotCountLimiter(), - } - e := echo.New() - - post := func(validator string, count uint32) (int, error) { - root, err := modelStore.GetRootAt(count) - assert.NoError(t, err) - reqBody := &PocProofsRequest{ - PocStageStartBlockHeight: 100, - ModelId: "model-a", - RootHash: base64.StdEncoding.EncodeToString(root), - Count: StringUint32(count), - LeafIndices: []StringUint32{0}, - ValidatorAddress: validator, - ValidatorSignerAddress: validator, - Timestamp: StringInt64(time.Now().UnixNano()), - } - signature, err := privKey.Sign(buildPocProofsSignPayload(reqBody, root)) - assert.NoError(t, err) - reqBody.Signature = base64.StdEncoding.EncodeToString(signature) - - body, err := json.Marshal(reqBody) - assert.NoError(t, err) - req := httptest.NewRequest(http.MethodPost, "/v1/poc/proofs", bytes.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - rec := httptest.NewRecorder() - if err := server.postPocProofs(e.NewContext(req, rec)); err != nil { - var httpErr *echo.HTTPError - if assert.ErrorAs(t, err, &httpErr) { - return httpErr.Code, err - } - return 0, err - } - return rec.Code, nil - } - - // First three distinct snapshot counts are served. - for count := uint32(1); count <= 3; count++ { - code, err := post("validator-address", count) - assert.NoError(t, err) - assert.Equal(t, http.StatusOK, code) - } - - // Fourth distinct count trips the quota, even though the snapshot is valid. - code, _ := post("validator-address", 4) - assert.Equal(t, http.StatusTooManyRequests, code) - - // Repeats of an already-seen count are still served. - code, err = post("validator-address", 2) - assert.NoError(t, err) - assert.Equal(t, http.StatusOK, code) - - // A different validator has an independent quota. - code, err = post("other-validator", 4) - assert.NoError(t, err) - assert.Equal(t, http.StatusOK, code) -} diff --git a/decentralized-api/internal/server/public/poc_snapshot_limiter.go b/decentralized-api/internal/server/public/poc_snapshot_limiter.go deleted file mode 100644 index 7403e2f4c..000000000 --- a/decentralized-api/internal/server/public/poc_snapshot_limiter.go +++ /dev/null @@ -1,103 +0,0 @@ -package public - -import ( - "fmt" - "sync" - "time" -) - -const ( - // maxDistinctSnapshotCountsPerValidator caps how many distinct snapshot - // counts one validator may request proofs for, per (stage, model) store. - // Honest validation needs exactly two: the early checkpoint count and the - // final commitment count. The third slot absorbs a capture that straddled - // a commit boundary. Repeat requests for an already-seen count are always - // allowed (they are snapshot-cache hits and cost nothing to serve). - maxDistinctSnapshotCountsPerValidator = 3 - - // snapshotLimiterIdleTTL evicts idle per-validator entries. It comfortably - // covers the validation retry window (~14 minutes). - snapshotLimiterIdleTTL = 2 * time.Hour - - // snapshotLimiterMaxEntries bounds limiter memory against many validators - // and stages. Oldest entries are evicted first. - snapshotLimiterMaxEntries = 8192 -) - -type snapshotLimiterEntry struct { - counts map[uint32]struct{} - lastSeen time.Time -} - -// snapshotCountLimiter tracks the distinct snapshot counts each validator has -// requested per (stage, model) store. It exists to stop a registered validator -// from forcing repeated snapshot-tree rebuilds by cycling through many valid -// flush-boundary counts. It deliberately does not limit request volume for -// counts already seen: those are served from the snapshot cache. -type snapshotCountLimiter struct { - mu sync.Mutex - entries map[string]*snapshotLimiterEntry - max int - ttl time.Duration - now func() time.Time -} - -func newSnapshotCountLimiter() *snapshotCountLimiter { - return &snapshotCountLimiter{ - entries: make(map[string]*snapshotLimiterEntry), - max: maxDistinctSnapshotCountsPerValidator, - ttl: snapshotLimiterIdleTTL, - now: time.Now, - } -} - -// Allow reports whether the validator may request proofs against the given -// snapshot count of the (stage, model) store, recording the count if allowed. -// distinct is the number of distinct counts the validator has used after this -// request (including the rejected one when allowed is false), so callers can -// surface validators approaching the quota before it trips. -func (l *snapshotCountLimiter) Allow(validatorAddress string, stageHeight int64, modelID string, count uint32) (allowed bool, distinct int) { - key := fmt.Sprintf("%s|%d|%s", validatorAddress, stageHeight, modelID) - - l.mu.Lock() - defer l.mu.Unlock() - - now := l.now() - l.pruneLocked(now) - - entry, ok := l.entries[key] - if !ok { - entry = &snapshotLimiterEntry{counts: make(map[uint32]struct{}, l.max)} - l.entries[key] = entry - } - entry.lastSeen = now - - if _, seen := entry.counts[count]; seen { - return true, len(entry.counts) - } - if len(entry.counts) >= l.max { - return false, len(entry.counts) + 1 - } - entry.counts[count] = struct{}{} - return true, len(entry.counts) -} - -func (l *snapshotCountLimiter) pruneLocked(now time.Time) { - for key, entry := range l.entries { - if now.Sub(entry.lastSeen) > l.ttl { - delete(l.entries, key) - } - } - // Hard cap as a memory backstop; evict oldest entries first. - for len(l.entries) >= snapshotLimiterMaxEntries { - var oldestKey string - var oldest time.Time - for key, entry := range l.entries { - if oldestKey == "" || entry.lastSeen.Before(oldest) { - oldestKey = key - oldest = entry.lastSeen - } - } - delete(l.entries, oldestKey) - } -} diff --git a/decentralized-api/internal/server/public/poc_snapshot_limiter_test.go b/decentralized-api/internal/server/public/poc_snapshot_limiter_test.go deleted file mode 100644 index 61d729199..000000000 --- a/decentralized-api/internal/server/public/poc_snapshot_limiter_test.go +++ /dev/null @@ -1,71 +0,0 @@ -package public - -import ( - "testing" - "time" -) - -func TestSnapshotCountLimiter(t *testing.T) { - t.Run("caps distinct counts, allows repeats", func(t *testing.T) { - l := newSnapshotCountLimiter() - - for i, count := range []uint32{100, 200, 300} { - allowed, distinct := l.Allow("val1", 1000, "m1", count) - if !allowed { - t.Fatalf("distinct count %d (#%d) should be allowed", count, i+1) - } - if distinct != i+1 { - t.Fatalf("distinct = %d, want %d", distinct, i+1) - } - } - if allowed, distinct := l.Allow("val1", 1000, "m1", 400); allowed { - t.Fatal("4th distinct count should be rejected") - } else if distinct != 4 { - t.Fatalf("rejected distinct = %d, want 4", distinct) - } - // Repeats of already-seen counts stay allowed. - for _, count := range []uint32{100, 200, 300} { - if allowed, _ := l.Allow("val1", 1000, "m1", count); !allowed { - t.Fatalf("repeat of count %d should be allowed", count) - } - } - // Still rejected after repeats. - if allowed, _ := l.Allow("val1", 1000, "m1", 500); allowed { - t.Fatal("new distinct count should stay rejected") - } - }) - - t.Run("quota is per validator, stage, and model", func(t *testing.T) { - l := newSnapshotCountLimiter() - for _, count := range []uint32{1, 2, 3} { - l.Allow("val1", 1000, "m1", count) - } - if allowed, _ := l.Allow("val2", 1000, "m1", 4); !allowed { - t.Fatal("other validator must have its own quota") - } - if allowed, _ := l.Allow("val1", 2000, "m1", 4); !allowed { - t.Fatal("other stage must have its own quota") - } - if allowed, _ := l.Allow("val1", 1000, "m2", 4); !allowed { - t.Fatal("other model must have its own quota") - } - }) - - t.Run("idle entries expire", func(t *testing.T) { - l := newSnapshotCountLimiter() - current := time.Unix(0, 0) - l.now = func() time.Time { return current } - - for _, count := range []uint32{1, 2, 3} { - l.Allow("val1", 1000, "m1", count) - } - if allowed, _ := l.Allow("val1", 1000, "m1", 4); allowed { - t.Fatal("quota should be exhausted") - } - - current = current.Add(snapshotLimiterIdleTTL + time.Minute) - if allowed, _ := l.Allow("val1", 1000, "m1", 4); !allowed { - t.Fatal("expired entry should reset the quota") - } - }) -} diff --git a/decentralized-api/internal/server/public/server.go b/decentralized-api/internal/server/public/server.go index 06666c5af..6abc72f8b 100644 --- a/decentralized-api/internal/server/public/server.go +++ b/decentralized-api/internal/server/public/server.go @@ -41,7 +41,6 @@ type Server struct { authzCache *authzcache.AuthzCache httpClient *http.Client statsStorage statsstorage.StatsStorage - pocSnapshotLimiter *snapshotCountLimiter } // ServerOption configures optional Server dependencies. @@ -90,7 +89,6 @@ func NewServer( epochGroupDataCache: internal.NewEpochGroupDataCache(recorder), authzCache: authzcache.NewAuthzCache(recorder), httpClient: NewNoRedirectClient(httpClientTimeout), - pocSnapshotLimiter: newSnapshotCountLimiter(), } for _, opt := range opts { diff --git a/decentralized-api/poc/artifacts/smst.go b/decentralized-api/poc/artifacts/smst.go index 263fba0ea..59d50f47e 100644 --- a/decentralized-api/poc/artifacts/smst.go +++ b/decentralized-api/poc/artifacts/smst.go @@ -3,6 +3,8 @@ package artifacts import ( "crypto/sha256" "encoding/binary" + "runtime" + "sync" ) const ( @@ -13,6 +15,9 @@ const ( // committed roots and must only happen at a coordinated protocol upgrade. smstDefaultDepth = 24 smstMaxDepth = 32 + // smstParallelHashMinCount: only fan out ensureHashed work when both + // children need hashing and the subtree has at least this many leaves. + smstParallelHashMinCount = 256 ) type smstNode struct { @@ -31,6 +36,20 @@ type SMST struct { emptyHash [][]byte leafCount uint32 hasNonce map[int32]bool // tracks which nonces exist (for duplicate detection) + + // navExistence makes HasNonce walk the tree instead of the hasNonce map. + // Snapshot views share nodes with the live tree but carry no map, so nonce + // existence must be read from the retained structure at that count. + navExistence bool + + // deferredHash invalidates node.hash on insert and fills via ensureHashed + // (default). When false, hashes are computed on every insert (upgrade-v0.2.14). + deferredHash bool + + // parallelHash fans out ensureHashed across GOMAXPROCS when both children + // need hashing (default on). Eager per-insert path hashing stays serial + // (parent depends on child); this flag mainly accelerates deferred fill. + parallelHash bool } // NewSMST creates a new sparse merkle sum tree. @@ -44,9 +63,11 @@ func NewSMST(depth int) *SMST { } s := &SMST{ - depth: depth, - emptyHash: make([][]byte, depth+1), - hasNonce: make(map[int32]bool), + depth: depth, + emptyHash: make([][]byte, depth+1), + hasNonce: make(map[int32]bool), + deferredHash: true, + parallelHash: true, } s.emptyHash[0] = smstHashEmpty() @@ -99,7 +120,11 @@ func (s *SMST) insertAt(node *smstNode, path []bool, level int, leafHash []byte) } node.count = s.nodeCount(node.left) + s.nodeCount(node.right) - node.hash = s.computeHash(node, level) + if s.deferredHash { + node.hash = nil // invalidated; recomputed lazily by ensureHashed + } else { + node.hash = s.computeHash(node, level) + } return node } @@ -129,9 +154,73 @@ func (s *SMST) GetRoot() ([]byte, uint32) { if s.root == nil { return s.emptyHash[s.depth], 0 } + s.ensureHashed() return s.root.hash, s.root.count } +// ensureHashed fills node hashes left nil by insertAt. Each node is hashed once +// here rather than on every insert that passes through it, so a shared upper +// node is not re-hashed for each descendant insert. Idempotent: a node whose +// hash is already set stops the recursion. When parallelHash is on, independent +// subtrees are hashed concurrently (bounded by GOMAXPROCS). +func (s *SMST) ensureHashed() { + if s.root == nil { + return + } + if !s.parallelHash { + s.hashNode(s.root, 0) + return + } + workers := runtime.GOMAXPROCS(0) + if workers < 2 { + s.hashNode(s.root, 0) + return + } + // Main goroutine counts as one worker; sem holds the remaining slots. + sem := make(chan struct{}, workers-1) + s.hashNodeParallel(s.root, 0, sem) +} + +func (s *SMST) hashNode(node *smstNode, level int) { + if node == nil || node.hash != nil { + return + } + s.hashNode(node.left, level+1) + s.hashNode(node.right, level+1) + node.hash = s.computeHash(node, level) +} + +func (s *SMST) hashNodeParallel(node *smstNode, level int, sem chan struct{}) { + if node == nil || node.hash != nil { + return + } + left, right := node.left, node.right + canFanOut := left != nil && left.hash == nil && + right != nil && right.hash == nil && + node.count >= smstParallelHashMinCount + if canFanOut { + select { + case sem <- struct{}{}: + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + defer func() { <-sem }() + s.hashNodeParallel(left, level+1, sem) + }() + s.hashNodeParallel(right, level+1, sem) + wg.Wait() + node.hash = s.computeHash(node, level) + return + default: + // All worker slots busy; finish this subtree on the current goroutine. + } + } + s.hashNodeParallel(left, level+1, sem) + s.hashNodeParallel(right, level+1, sem) + node.hash = s.computeHash(node, level) +} + // GetLeafByDenseIndex navigates to a leaf using sum-based dense indexing. // Returns the nonce and proof (sibling hashes from root to leaf). func (s *SMST) GetLeafByDenseIndex(denseIndex uint32) (int32, [][]byte, error) { @@ -194,9 +283,36 @@ func (s *SMST) Depth() int { // HasNonce checks if a nonce exists in the tree. func (s *SMST) HasNonce(nonce int32) bool { + if s.navExistence { + return s.hasNonceInTree(nonce) + } return s.hasNonce[nonce] } +func (s *SMST) hasNonceInTree(nonce int32) bool { + if s.root == nil { + return false + } + // A nonce needing more depth than this tree has could never have been + // inserted without expanding it, so it is absent. Below the depth the path + // bits are injective, making the walk exact — matching the hasNonce map. + if s.requiredDepth(nonce) > s.depth { + return false + } + node := s.root + for _, goRight := range s.noncePath(nonce) { + if node == nil { + return false + } + if goRight { + node = node.right + } else { + node = node.left + } + } + return node != nil +} + func (s *SMST) denseIndexForNonce(nonce int32) (uint32, error) { if s.root == nil || !s.HasNonce(nonce) { return 0, ErrNonceNotFound @@ -270,15 +386,24 @@ func (s *SMST) expandDepth(newDepth int) { diff := newDepth - oldDepth for i := 0; i < diff; i++ { if s.root != nil { - // This wrapper will be at level (diff - 1 - i) in final tree - level := diff - 1 - i - siblingHeight := newDepth - level - 1 - newRoot := &smstNode{ - left: s.root, - count: s.root.count, + if s.deferredHash { + // Wrapper hash is deferred; ensureHashed fills it from the wrapped + // subtree and the empty sibling, identical to eager computation. + s.root = &smstNode{ + left: s.root, + count: s.root.count, + } + } else { + // This wrapper will be at level (diff - 1 - i) in final tree + level := diff - 1 - i + siblingHeight := newDepth - level - 1 + newRoot := &smstNode{ + left: s.root, + count: s.root.count, + } + newRoot.hash = smstHashNode(s.root.hash, s.emptyHash[siblingHeight], newRoot.count) + s.root = newRoot } - newRoot.hash = smstHashNode(s.root.hash, s.emptyHash[siblingHeight], newRoot.count) - s.root = newRoot } } } diff --git a/decentralized-api/poc/artifacts/smst_cow.go b/decentralized-api/poc/artifacts/smst_cow.go new file mode 100644 index 000000000..493b3713d --- /dev/null +++ b/decentralized-api/poc/artifacts/smst_cow.go @@ -0,0 +1,165 @@ +package artifacts + +import ( + "os" + "strconv" + "strings" +) + +// SMST_COW / SMST_DEFERRED_HASH / SMST_SNAPSHOT_IN_MEMORY_CLONE / +// SMST_PARALLEL_HASH default on when unset. +// Profiling overrides: +// +// SMST_COW=0 — in-place Insert (no path-copy) +// SMST_DEFERRED_HASH=0 — hash on every insert (upgrade-v0.2.14 baseline) +// SMST_SNAPSHOT_IN_MEMORY_CLONE=0 — tip Prebuild rebuilds from artifacts without holding +// the write lock (upgrade-v0.2.14 Warm/Prebuild path); +// default 1 = deep in-memory clone under write lock +// SMST_PARALLEL_HASH=0 — serial ensureHashed; default 1 = multicore fill +const envSMSTCOW = "SMST_COW" +const envSMSTDeferredHash = "SMST_DEFERRED_HASH" +const envSMSTSnapshotInMemoryClone = "SMST_SNAPSHOT_IN_MEMORY_CLONE" +const envSMSTParallelHash = "SMST_PARALLEL_HASH" + +func smstEnvBool(key string, def bool) bool { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return def + } + switch strings.ToLower(v) { + case "0", "false", "off", "no": + return false + case "1", "true", "on", "yes": + return true + } + b, err := strconv.ParseBool(v) + if err != nil { + return def + } + return b +} + +// smstCOWEnabledFromEnv reports COW inserts; default true. +func smstCOWEnabledFromEnv() bool { + return smstEnvBool(envSMSTCOW, true) +} + +// smstDeferredHashFromEnv reports deferred Merkle hashing; default true. +func smstDeferredHashFromEnv() bool { + return smstEnvBool(envSMSTDeferredHash, true) +} + +// smstSnapshotInMemoryCloneFromEnv reports tip-snapshot strategy when COW is off: +// true = deep clone under write lock; false = artifact rebuild without write lock. +// Default true. Ignored when COW retains O(1) at flush. +func smstSnapshotInMemoryCloneFromEnv() bool { + return smstEnvBool(envSMSTSnapshotInMemoryClone, true) +} + +// smstParallelHashFromEnv reports multicore ensureHashed; default true. +func smstParallelHashFromEnv() bool { + return smstEnvBool(envSMSTParallelHash, true) +} + +// smstSnapshot is an immutable capture of the tree at a specific leaf count. +// insertCOW never mutates existing nodes, so a captured root stays valid for the +// life of the tree and serves proofs without any rebuild. It is captured after a +// GetRoot (flush/recover), so its nodes are already hashed. +type smstSnapshot struct { + root *smstNode + depth int + count uint32 +} + +// insertCOW is a copy-on-write Insert: it rewrites only the nodes on the +// root->leaf path and shares every untouched sibling subtree, so prior roots are +// never clobbered — which is what makes snapshots free. Roots and counts are +// byte-identical to Insert; hashing is deferred identically (filled by +// ensureHashed at GetRoot). +func (s *SMST) insertCOW(nonce int32, leafHash []byte) (uint32, error) { + if s.hasNonce[nonce] { + return 0, ErrDuplicateNonce + } + + requiredDepth := s.requiredDepth(nonce) + if requiredDepth > s.depth { + s.expandDepth(requiredDepth) + } + + path := s.noncePath(nonce) + s.root = s.insertAtCOW(s.root, path, 0, leafHash) + + s.hasNonce[nonce] = true + s.leafCount++ + + return s.leafCount, nil +} + +func (s *SMST) insertAtCOW(node *smstNode, path []bool, level int, leafHash []byte) *smstNode { + if level == s.depth { + return &smstNode{hash: leafHash, count: 1} + } + + newNode := &smstNode{} + if node != nil { + newNode.left = node.left + newNode.right = node.right + } + + if path[level] { + newNode.right = s.insertAtCOW(newNode.right, path, level+1, leafHash) + } else { + newNode.left = s.insertAtCOW(newNode.left, path, level+1, leafHash) + } + + newNode.count = s.nodeCount(newNode.left) + s.nodeCount(newNode.right) + if s.deferredHash { + newNode.hash = nil // deferred; filled by ensureHashed, identical to insertAt + } else { + newNode.hash = s.computeHash(newNode, level) + } + + return newNode +} + +// snapshot captures the current tree. O(1): it retains the root pointer and the +// depth in force at this count (the depth a historical proof must use). Callers +// capture after GetRoot, so the retained nodes are already hashed. +func (s *SMST) snapshot() smstSnapshot { + return smstSnapshot{root: s.root, depth: s.depth, count: s.leafCount} +} + +// cloneSnapshot deep-copies the live tree into an independent snapshot. Used when +// COW is disabled so later in-place inserts cannot mutate historical nodes. +// Callers must hold the store write lock and have already ensureHashed. +func (s *SMST) cloneSnapshot() smstSnapshot { + return smstSnapshot{root: cloneSMSTNode(s.root), depth: s.depth, count: s.leafCount} +} + +func cloneSMSTNode(n *smstNode) *smstNode { + if n == nil { + return nil + } + out := &smstNode{count: n.count} + if n.hash != nil { + out.hash = append([]byte(nil), n.hash...) + } + out.left = cloneSMSTNode(n.left) + out.right = cloneSMSTNode(n.right) + return out +} + +// snapshotView returns a read-only SMST bound to a captured snapshot so proof +// serving uses the depth that was in force at that count, not the live depth. +// emptyHash indices 0..depth are stable across expansion (append-only), so the +// live table is safe to share. Existence is read from the retained structure +// (navExistence) since a snapshot carries no per-count nonce map. +func (s *SMST) snapshotView(snap smstSnapshot) *SMST { + return &SMST{ + root: snap.root, + depth: snap.depth, + emptyHash: s.emptyHash, + leafCount: snap.count, + navExistence: true, + } +} diff --git a/decentralized-api/poc/artifacts/smst_cow_store_test.go b/decentralized-api/poc/artifacts/smst_cow_store_test.go new file mode 100644 index 000000000..ea69d59a5 --- /dev/null +++ b/decentralized-api/poc/artifacts/smst_cow_store_test.go @@ -0,0 +1,505 @@ +package artifacts + +import ( + "bytes" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestCOWStoreProofPaths verifies proofs served through the copy-on-write store +// across the three paths: the live tip, a historical committed count served from +// a retained snapshot (O(depth), no rebuild), and the same historical count +// after a restart (retained is in-memory, so recovery re-captures only the tip +// and older counts fall back to the exact rebuild). All must verify against the +// count's root. +func TestCOWStoreProofPaths(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + + // Fill and flush in blocks so several committed counts exist. + const block = 2000 + const blocks = 4 + var earlyCount uint32 + for b := 0; b < blocks; b++ { + for i := b * block; i < (b+1)*block; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush: %v", err) + } + if b == 0 { + earlyCount = store.Count() // first committed count + } + } + + if len(store.retained) == 0 { + t.Fatalf("no retained snapshots captured") + } + + verifyAt := func(tag string, count uint32) { + root, err := store.GetRootAt(count) + if err != nil { + t.Fatalf("%s GetRootAt(%d): %v", tag, count, err) + } + entries, err := store.GetArtifactsAndProofs([]uint32{count / 2}, count) + if err != nil { + t.Fatalf("%s proof@%d: %v", tag, count, err) + } + e := entries[0] + if !VerifySMSTProofSlice(root, count, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("%s proof did not verify at count %d", tag, count) + } + } + + // Live tip and a historical committed count (retained snapshot path). + verifyAt("live", store.Count()) + verifyAt("retained-historical", earlyCount) + + if err := store.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + // Restart: recovery rebuilds the live tree and re-captures the tip; an older + // committed count falls back to the exact rebuild and must still verify. + store2, err := OpenSMST(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer store2.Close() + if store2.Count() != uint32(block*blocks) { + t.Fatalf("recovered count %d, want %d", store2.Count(), block*blocks) + } + // Recovery must re-capture retained snapshots at every committed count, not + // just the tip, so historical proofs are served in O(depth) after a restart. + if len(store2.retained) < blocks { + t.Fatalf("post-restart retained=%d, want >= %d committed counts", len(store2.retained), blocks) + } + if _, ok := store2.retained[earlyCount]; !ok { + t.Fatalf("early committed count %d not re-captured after restart", earlyCount) + } + + root, err := store2.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("post-restart GetRootAt(%d): %v", earlyCount, err) + } + entries, err := store2.GetArtifactsAndProofs([]uint32{earlyCount / 2}, earlyCount) + if err != nil { + t.Fatalf("post-restart proof@%d: %v", earlyCount, err) + } + e := entries[0] + if !VerifySMSTProofSlice(root, earlyCount, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("post-restart historical proof did not verify") + } +} + +// TestCOWRetainedProofsDoNotHoldWriteLock checks that historical retained proofs +// release the write lock before artifact I/O: concurrent proof readers at an +// early committed count must not serialize ingest. Under -race this also guards +// the unlock→RLock handoff in acquireSnapshotTree. +func TestCOWRetainedProofsDoNotHoldWriteLock(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + + const early = 2000 + for i := 0; i < early; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush: %v", err) + } + earlyCount := store.Count() + earlyRoot, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("GetRootAt(%d): %v", earlyCount, err) + } + if _, ok := store.retained[earlyCount]; !ok { + t.Fatalf("expected retained snapshot at %d", earlyCount) + } + + var ( + wg sync.WaitGroup + proofOK int64 + proofErr int64 + writesOK int64 + readerDone = make(chan struct{}) + nextNonce = int32(early) + ) + + const readers = 8 + for g := 0; g < readers; g++ { + wg.Add(1) + go func(gid int) { + defer wg.Done() + idx := uint32(gid) % earlyCount + for { + select { + case <-readerDone: + return + default: + } + entries, err := store.GetArtifactsAndProofs([]uint32{idx}, earlyCount) + if err != nil || len(entries) != 1 { + atomic.AddInt64(&proofErr, 1) + continue + } + e := entries[0] + if !VerifySMSTProofSlice(earlyRoot, earlyCount, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + atomic.AddInt64(&proofErr, 1) + continue + } + atomic.AddInt64(&proofOK, 1) + } + }(g) + } + + // Writer must make progress while retained proofs are in flight. If proofs + // still held the write lock across I/O, Adds would stall behind every batch. + deadline := time.After(500 * time.Millisecond) + for { + select { + case <-deadline: + close(readerDone) + wg.Wait() + if atomic.LoadInt64(&proofOK) == 0 { + t.Fatalf("no successful retained proofs") + } + if atomic.LoadInt64(&proofErr) != 0 { + t.Fatalf("retained proof errors: %d", proofErr) + } + if atomic.LoadInt64(&writesOK) == 0 { + t.Fatalf("ingest made no progress while retained proofs ran (write lock held across I/O?)") + } + return + default: + n := atomic.AddInt32(&nextNonce, 1) - 1 + if err := store.AddWithNode(n, testVector(int(n)), "n"); err != nil { + t.Fatalf("concurrent add %d: %v", n, err) + } + if n%200 == 0 { + if err := store.Flush(); err != nil { + t.Fatalf("concurrent flush: %v", err) + } + } + atomic.AddInt64(&writesOK, 1) + } + } +} + +// TestFlushedRootsSurviveLostDistributionHistory checks that a flush count +// remains provable after restart even when distributions.jsonl lost that entry +// (the warn-only appendDistributionSnapshot failure mode). flushed_roots.jsonl +// is the durable commit journal used to re-capture retained snapshots. +func TestFlushedRootsSurviveLostDistributionHistory(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + + const early = 500 + for i := 0; i < early; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush early: %v", err) + } + earlyCount := store.Count() + earlyRoot, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("GetRootAt(%d): %v", earlyCount, err) + } + earlyRoot = bytes.Clone(earlyRoot) + + for i := early; i < early*2; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush final: %v", err) + } + if err := store.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + // Simulate distribution append failure / loss: wipe distributions.jsonl but + // keep flushed_roots.jsonl (and artifacts.data). + if err := os.Truncate(filepath.Join(dir, "distributions.jsonl"), 0); err != nil { + t.Fatalf("truncate distributions: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "flushed_roots.jsonl")); err != nil { + t.Fatalf("flushed_roots.jsonl missing after flush: %v", err) + } + + store2, err := OpenSMST(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer store2.Close() + + if _, ok := store2.distributionHistory[earlyCount]; ok { + t.Fatalf("expected distributionHistory to lack %d after truncate", earlyCount) + } + if _, ok := store2.retained[earlyCount]; !ok { + t.Fatalf("expected retained snapshot at %d after restart without dist history", earlyCount) + } + root2, err := store2.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("post-restart GetRootAt(%d): %v", earlyCount, err) + } + if !bytes.Equal(earlyRoot, root2) { + t.Fatalf("early root changed across restart without dist history") + } + entries, err := store2.GetArtifactsAndProofs([]uint32{earlyCount / 2}, earlyCount) + if err != nil { + t.Fatalf("post-restart proof@%d: %v", earlyCount, err) + } + e := entries[0] + if !VerifySMSTProofSlice(root2, earlyCount, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("post-restart early proof did not verify") + } +} + +// TestSMSTCOWEnvDisabled uses in-place Insert (deferred hashing only): no +// retained snapshots, roots still match the COW path. +func TestSMSTCOWEnvDisabled(t *testing.T) { + t.Setenv(envSMSTCOW, "0") + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + if store.cowEnabled { + t.Fatalf("expected cowEnabled=false with %s=0", envSMSTCOW) + } + + const n = 500 + for i := 0; i < n; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush: %v", err) + } + if len(store.retained) != 0 { + t.Fatalf("retained=%d, want 0 when COW disabled", len(store.retained)) + } + root := store.GetRoot() + if root == nil { + t.Fatal("nil root") + } + + // Same workload with COW on must produce the same tip root. + t.Setenv(envSMSTCOW, "1") + dir2 := t.TempDir() + store2, err := OpenSMST(dir2) + if err != nil { + t.Fatalf("OpenSMST cow: %v", err) + } + defer store2.Close() + if !store2.cowEnabled { + t.Fatal("expected cowEnabled=true") + } + for i := 0; i < n; i++ { + if err := store2.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("cow add %d: %v", i, err) + } + } + if err := store2.Flush(); err != nil { + t.Fatalf("cow flush: %v", err) + } + root2 := store2.GetRoot() + if !bytes.Equal(root, root2) { + t.Fatalf("root mismatch COW off vs on:\n off=%x\n on= %x", root, root2) + } +} + +// TestSMSTCOWDisabledEarlyDeepClone checks that with SMST_COW=0 and +// SMST_SNAPSHOT_IN_MEMORY_CLONE=1 (default), PrebuildSnapshot deep-clones the live tip +// under the write lock into retained so proofs stay valid after the tip advances. +func TestSMSTCOWDisabledEarlyDeepClone(t *testing.T) { + t.Setenv(envSMSTCOW, "0") + t.Setenv(envSMSTSnapshotInMemoryClone, "1") + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + if !store.snapshotInMemoryClone { + t.Fatal("expected snapshotInMemoryClone=true") + } + + const early = 300 + for i := 0; i < early; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush early: %v", err) + } + earlyCount := store.Count() + earlyRoot, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("GetRootAt(%d): %v", earlyCount, err) + } + earlyRoot = bytes.Clone(earlyRoot) + + // Commit-worker style: deep clone while tip still equals early count. + if err := store.PrebuildSnapshot(earlyCount); err != nil { + t.Fatalf("PrebuildSnapshot(%d): %v", earlyCount, err) + } + if _, ok := store.retained[earlyCount]; !ok { + t.Fatalf("expected deep-cloned retained snapshot at %d", earlyCount) + } + globalSnapshotCache.mu.Lock() + _, cacheHit := globalSnapshotCache.entries[snapshotCacheKey{store: store, count: earlyCount}] + globalSnapshotCache.mu.Unlock() + if cacheHit { + t.Fatalf("tip Prebuild must not rebuild into snapshot cache when deep-cloning") + } + + // Advance tip past the early commit (in-place Insert mutates live nodes). + for i := early; i < early*2; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush final: %v", err) + } + + root2, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("post-advance GetRootAt(%d): %v", earlyCount, err) + } + if !bytes.Equal(earlyRoot, root2) { + t.Fatalf("early root changed after tip advanced") + } + entries, err := store.GetArtifactsAndProofs([]uint32{earlyCount / 2}, earlyCount) + if err != nil { + t.Fatalf("early proof after tip advanced: %v", err) + } + e := entries[0] + if !VerifySMSTProofSlice(root2, earlyCount, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("early proof did not verify after tip advanced") + } +} + +// TestSMSTCOWDisabledEarlyArtifactRebuild checks SMST_SNAPSHOT_IN_MEMORY_CLONE=0: +// tip Prebuild rebuilds from artifacts into the process cache without retaining +// a deep clone (upgrade-v0.2.14 path). +func TestSMSTCOWDisabledEarlyArtifactRebuild(t *testing.T) { + t.Setenv(envSMSTCOW, "0") + t.Setenv(envSMSTSnapshotInMemoryClone, "0") + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + if store.snapshotInMemoryClone { + t.Fatal("expected snapshotInMemoryClone=false") + } + + const early = 300 + for i := 0; i < early; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush early: %v", err) + } + earlyCount := store.Count() + earlyRoot, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("GetRootAt(%d): %v", earlyCount, err) + } + earlyRoot = bytes.Clone(earlyRoot) + + if err := store.PrebuildSnapshot(earlyCount); err != nil { + t.Fatalf("PrebuildSnapshot(%d): %v", earlyCount, err) + } + if len(store.retained) != 0 { + t.Fatalf("retained must stay empty with snapshot clone off, got %d", len(store.retained)) + } + globalSnapshotCache.mu.Lock() + entry, ok := globalSnapshotCache.entries[snapshotCacheKey{store: store, count: earlyCount}] + globalSnapshotCache.mu.Unlock() + if !ok || entry == nil || !entry.pinned { + t.Fatalf("expected pinned snapshot-cache rebuild at %d", earlyCount) + } + + for i := early; i < early*2; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush final: %v", err) + } + + root2, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("post-advance GetRootAt(%d): %v", earlyCount, err) + } + if !bytes.Equal(earlyRoot, root2) { + t.Fatalf("early root changed after tip advanced") + } + entries, err := store.GetArtifactsAndProofs([]uint32{earlyCount / 2}, earlyCount) + if err != nil { + t.Fatalf("early proof after tip advanced: %v", err) + } + e := entries[0] + if !VerifySMSTProofSlice(root2, earlyCount, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("early proof did not verify after tip advanced") + } +} + +// TestSMSTDefaultsDeferredAndCOW ensures unset env keeps production defaults: +// deferred hashing on, COW on, tip snapshot clone on, parallel hash on. +func TestSMSTDefaultsDeferredAndCOW(t *testing.T) { + t.Setenv(envSMSTCOW, "") + t.Setenv(envSMSTDeferredHash, "") + t.Setenv(envSMSTSnapshotInMemoryClone, "") + t.Setenv(envSMSTParallelHash, "") + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + if !store.cowEnabled { + t.Fatal("cowEnabled default want true") + } + if !store.smst.deferredHash { + t.Fatal("deferredHash default want true") + } + if !store.snapshotInMemoryClone { + t.Fatal("snapshotInMemoryClone default want true") + } + if !store.smst.parallelHash { + t.Fatal("parallelHash default want true") + } +} diff --git a/decentralized-api/poc/artifacts/smst_deferred_test.go b/decentralized-api/poc/artifacts/smst_deferred_test.go new file mode 100644 index 000000000..fcff4678b --- /dev/null +++ b/decentralized-api/poc/artifacts/smst_deferred_test.go @@ -0,0 +1,379 @@ +package artifacts + +import ( + "bytes" + "crypto/sha256" + "encoding/binary" + "fmt" + "os" + "strconv" + "sync" + "testing" + "time" +) + +func testVector(i int) []byte { + v := make([]byte, 24) + binary.LittleEndian.PutUint64(v[0:8], uint64(i)) + binary.LittleEndian.PutUint64(v[8:16], uint64(i*2654435761)) + return v +} + +func perfN(def int) int { + if v := os.Getenv("SMST_PERF_N"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return def +} + +// deferredDataset returns a deterministic nonce sequence that forces early depth +// expansion (nonces >= 2^25), includes negatives (path via uint32, depth 32), +// then a long sequential run — exercising every hashing path a flush touches. +func deferredDataset(n int) []int32 { + out := make([]int32, 0, n+8) + out = append(out, 1<<25, (1<<25)+1, 1<<28, -1, -2, -1000000) + for i := 0; i < n; i++ { + out = append(out, int32(i)) + } + return out +} + +// TestDeferredHashFingerprint builds a large store with periodic flushes, +// verifies every sampled proof against its flush-count root, and folds all +// roots plus sampled proofs into a single deterministic SHA-256 fingerprint. +// The fingerprint is stable across runs; each proof is verified independently +// against its root, so a matching fingerprint means deferred hashing reproduces +// the same roots and proofs the tree commits to. +func TestDeferredHashFingerprint(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + + nonces := deferredDataset(200000) + fp := sha256.New() + const flushEvery = 20000 + added := 0 + + for _, nonce := range nonces { + if err := store.AddWithNode(nonce, testVector(int(nonce)), "n"); err != nil { + t.Fatalf("add %d: %v", nonce, err) + } + added++ + if added%flushEvery != 0 { + continue + } + if err := store.Flush(); err != nil { + t.Fatalf("flush: %v", err) + } + count := store.Count() + root, err := store.GetRootAt(count) + if err != nil { + t.Fatalf("GetRootAt(%d): %v", count, err) + } + var cb [4]byte + binary.LittleEndian.PutUint32(cb[:], count) + fp.Write(cb[:]) + fp.Write(root) + + for _, idx := range []uint32{0, count / 3, count / 2, count - 1} { + entries, err := store.GetArtifactsAndProofs([]uint32{idx}, count) + if err != nil { + t.Fatalf("proof idx=%d count=%d: %v", idx, count, err) + } + e := entries[0] + leaf := encodeLeaf(e.Nonce, e.Vector) + if !VerifySMSTProofSlice(root, count, e.Nonce, leaf, e.Proof) { + t.Fatalf("proof failed idx=%d count=%d nonce=%d", idx, count, e.Nonce) + } + for _, p := range e.Proof { + fp.Write(p) + } + } + } + + t.Logf("SMST FINGERPRINT (200k + specials, flush 20k): %x", fp.Sum(nil)) +} + +// TestDeferredGetRootStableAndIdempotent checks the deferred-hash lifecycle in +// isolation: GetRoot on an empty tree returns the empty root, and after inserts +// two consecutive GetRoot calls (fill-then-reuse) return the identical hash. +func TestDeferredGetRootStableAndIdempotent(t *testing.T) { + tree := NewSMST(0) + empty, count := tree.GetRoot() + if count != 0 || len(empty) == 0 { + t.Fatalf("empty tree: count=%d rootLen=%d", count, len(empty)) + } + + for i := 0; i < 5000; i++ { + if _, err := tree.Insert(int32(i), smstHashLeaf(testVector(i))); err != nil { + t.Fatalf("insert %d: %v", i, err) + } + } + first, c1 := tree.GetRoot() + second, c2 := tree.GetRoot() // no new inserts: ensureHashed must be a no-op + if c1 != c2 || c1 != 5000 { + t.Fatalf("count drift: %d vs %d", c1, c2) + } + if string(first) != string(second) { + t.Fatalf("root not stable across GetRoot calls") + } +} + +// TestParallelHashMatchesSerial checks multicore ensureHashed produces the same +// root as serial fill (and as eager inserts). +func TestParallelHashMatchesSerial(t *testing.T) { + const n = 20_000 + const stride = 100 + + build := func(deferred, parallel bool) []byte { + tree := NewSMST(0) + tree.deferredHash = deferred + tree.parallelHash = parallel + for i := 0; i < n; i++ { + if _, err := tree.Insert(int32(i*stride), smstHashLeaf(testVector(i))); err != nil { + t.Fatalf("insert %d: %v", i, err) + } + } + root, count := tree.GetRoot() + if count != n { + t.Fatalf("count=%d want %d", count, n) + } + return root + } + + serial := build(true, false) + parallel := build(true, true) + eager := build(false, false) + eagerPar := build(false, true) // parallel unused on eager path fill; root must still match + if !bytes.Equal(serial, parallel) { + t.Fatalf("deferred serial vs parallel root mismatch:\n serial=%x\n paral=%x", serial, parallel) + } + if !bytes.Equal(serial, eager) { + t.Fatalf("deferred vs eager root mismatch:\n def=%x\n eager=%x", serial, eager) + } + if !bytes.Equal(eager, eagerPar) { + t.Fatalf("eager serial vs parallel-flag root mismatch:\n a=%x\n b=%x", eager, eagerPar) + } +} + +// TestDeferredLiveTipProofSlowThenFast exercises both live-tip proof paths in +// acquireSnapshotTree: a proof requested before any GetRoot must fill hashes +// under the write lock, then retry onto the RLock fast path; a later proof at +// the same count is served entirely under RLock. Both must verify. +func TestDeferredLiveTipProofSlowThenFast(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + + const n = 3000 + for i := 0; i < n; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + count := store.Count() + + // Slow path: no GetRoot yet, so the live tip still carries deferred hashes. + slow, err := store.GetArtifactsAndProofs([]uint32{count / 2}, count) + if err != nil { + t.Fatalf("slow-path proof: %v", err) + } + root, err := store.GetRootAt(count) + if err != nil { + t.Fatalf("GetRootAt: %v", err) + } + e := slow[0] + if !VerifySMSTProofSlice(root, count, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("slow-path proof did not verify") + } + + // Fast path: hashes now filled, same live count served under the read lock. + fast, err := store.GetArtifactsAndProofs([]uint32{count / 2}, count) + if err != nil { + t.Fatalf("fast-path proof: %v", err) + } + f := fast[0] + if !VerifySMSTProofSlice(root, count, f.Nonce, encodeLeaf(f.Nonce, f.Vector), f.Proof) { + t.Fatalf("fast-path proof did not verify") + } +} + +// TestDeferredLiveTipConcurrentProofsBeforeHashFill hammers the live tip with +// concurrent proofs before any GetRoot. Hash fill must run once under Lock; +// proof I/O must not hold the write lock, so readers finish without serializing +// the whole batch on Lock. Under -race this also covers the ensureHashed→retry +// handoff. +func TestDeferredLiveTipConcurrentProofsBeforeHashFill(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + + const n = 4000 + for i := 0; i < n; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + count := store.Count() + + const readers = 16 + var wg sync.WaitGroup + errCh := make(chan error, readers) + for g := 0; g < readers; g++ { + wg.Add(1) + go func(gid int) { + defer wg.Done() + idx := uint32(gid) % count + entries, err := store.GetArtifactsAndProofs([]uint32{idx}, count) + if err != nil { + errCh <- err + return + } + root, err := store.GetRootAt(count) + if err != nil { + errCh <- err + return + } + e := entries[0] + if !VerifySMSTProofSlice(root, count, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + errCh <- fmt.Errorf("proof verify failed for goroutine %d", gid) + } + }(g) + } + wg.Wait() + close(errCh) + for err := range errCh { + t.Fatalf("concurrent tip proof: %v", err) + } +} + +// TestDeferredHistoricalRebuildProof forces the O(N) rebuild path: a proof at a +// past flush count is reconstructed from the log by rebuildTreeFromInputs, which +// must hash the fresh tree (ensureHashed) before serving. The proof must verify +// against the historical root. +func TestDeferredHistoricalRebuildProof(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + + for i := 0; i < 4000; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + if err := store.Flush(); err != nil { + t.Fatalf("flush: %v", err) + } + historical := store.Count() + histRoot, err := store.GetRootAt(historical) + if err != nil { + t.Fatalf("GetRootAt(historical): %v", err) + } + + // Advance past the flush so `historical` is no longer the live tip. + for i := 4000; i < 6000; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + + entries, err := store.GetArtifactsAndProofs([]uint32{historical / 2}, historical) + if err != nil { + t.Fatalf("historical proof: %v", err) + } + e := entries[0] + if !VerifySMSTProofSlice(histRoot, historical, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("historical (rebuilt) proof did not verify") + } +} + +// TestDeferredStoreEdgeCases covers the error and empty-tree branches of the +// deferred-hash read paths: an empty store roots to nil, out-of-range and +// zero counts are rejected, and a closed store returns ErrStoreClosed rather +// than touching the tree. +func TestDeferredStoreEdgeCases(t *testing.T) { + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + + if r := store.GetRoot(); r != nil { + t.Fatalf("empty store root = %x, want nil", r) + } + if r, err := store.GetRootAt(0); err != nil || r != nil { + t.Fatalf("GetRootAt(0) = (%x, %v), want (nil, nil)", r, err) + } + + for i := 0; i < 500; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + } + count := store.Count() + + if _, err := store.GetRootAt(count + 1); err == nil { + t.Fatalf("GetRootAt(count+1) should error") + } + // Proof beyond the live count reaches acquireSnapshotTree's range check. + if _, err := store.GetArtifactsAndProofs([]uint32{0}, count+1); err == nil { + t.Fatalf("proof at count+1 should error") + } + + // Directly exercise acquireSnapshotTree's slow-path range check: an + // over-count that is not the live tip falls through to the write-locked + // branch and must report the mismatch. + if _, _, err := store.acquireSnapshotTree(count + 2); err == nil { + t.Fatalf("acquireSnapshotTree(count+2) should error") + } + + if err := store.Close(); err != nil { + t.Fatalf("close: %v", err) + } + if _, err := store.GetRootAt(count); err != ErrStoreClosed { + t.Fatalf("closed GetRootAt err = %v, want ErrStoreClosed", err) + } + if _, _, err := store.acquireSnapshotTree(count); err != ErrStoreClosed { + t.Fatalf("closed acquireSnapshotTree err = %v, want ErrStoreClosed", err) + } +} + +// TestIngestPerfPortable times building an N-leaf tree via Insert plus one +// GetRoot (which fills deferred hashes). It reports the ingest cost that +// deferred hashing reduces by hashing shared upper nodes once at GetRoot rather +// than on every insert that passes through them. +func TestIngestPerfPortable(t *testing.T) { + n := perfN(200000) + nonces := make([]int32, n) + leaves := make([][]byte, n) + for i := 0; i < n; i++ { + nonces[i] = int32(i) + leaves[i] = smstHashLeaf(testVector(i)) + } + + t0 := time.Now() + tree := NewSMST(0) + for i := 0; i < n; i++ { + if _, err := tree.Insert(nonces[i], leaves[i]); err != nil { + t.Fatalf("insert: %v", err) + } + } + root, _ := tree.GetRoot() + elapsed := time.Since(t0) + + t.Logf("INGEST N=%d: %v root=%x", n, elapsed, root[:8]) +} diff --git a/decentralized-api/poc/artifacts/smst_findings_verification_test.go b/decentralized-api/poc/artifacts/smst_findings_verification_test.go new file mode 100644 index 000000000..f80d85f3b --- /dev/null +++ b/decentralized-api/poc/artifacts/smst_findings_verification_test.go @@ -0,0 +1,287 @@ +package artifacts + +// Verification of Dmytro's PoC-proof-serving load-test findings against the +// #1432 changes (copy-on-write retained snapshots + deferred hashing). Each test +// emits a machine-parseable "RESULT ..." line consumed by the before/after +// comparison. The mirror file smst_findings_base_test.go runs the same drivers +// on the base (upgrade-v0.2.14) tree. +// +// Env knobs (defaults keep `go test` fast; the reported numbers use the large +// values): SMST_FIND_N (leaves), SMST_FIND_FLUSH (flush interval), SMST_FIND_K +// (distinct attack counts). + +import ( + "bytes" + "os" + "strconv" + "testing" + "time" +) + +func findEnvInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return def +} + +// buildFlushedStore fills a store with n leaves, flushing every `flush` leaves, +// and returns the store plus the ordered list of committed (flush) counts. +func buildFlushedStore(t *testing.T, n, flush int) (*SMSTArtifactStore, []uint32) { + t.Helper() + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + var committed []uint32 + for i := 0; i < n; i++ { + if err := store.AddWithNode(int32(i), testVector(i), "n"); err != nil { + t.Fatalf("add %d: %v", i, err) + } + if (i+1)%flush == 0 { + if err := store.Flush(); err != nil { + t.Fatalf("flush at %d: %v", i+1, err) + } + committed = append(committed, store.Count()) + } + } + return store, committed +} + +// --------------------------------------------------------------------------- +// Differential byte-identity: Insert (eager, base semantics) vs insertCOW +// (deferred, #1432 live path) vs rebuildTreeFromInputs (fallback path). +// Closes the coverage gap: nothing previously asserted these produce +// byte-identical roots AND proofs. A divergence would false-strip an honest +// node if a base-version and a #1432 node ever coexisted in one network. +// --------------------------------------------------------------------------- +func TestDiff_ByteIdentity_InsertVsCOWVsRebuild(t *testing.T) { + // Ns chosen to cross several depth-expansion boundaries. + ns := []int{1, 2, 100, 999, 1000, 4096, 10000} + if !testing.Short() { + ns = append(ns, 50000, 100000) + } + maxN := ns[len(ns)-1] + + // Precompute the shared (nonce, leafHash) sequence once. + type leaf struct { + nonce int32 + leafHash []byte + } + leaves := make([]leaf, maxN) + for i := 0; i < maxN; i++ { + nonce := int32(i) + leaves[i] = leaf{nonce: nonce, leafHash: smstHashLeaf(encodeLeaf(nonce, testVector(i)))} + } + + for _, n := range ns { + // Eager Insert tree (base semantics). + eager := NewSMST(smstDefaultDepth) + for i := 0; i < n; i++ { + if _, err := eager.Insert(leaves[i].nonce, leaves[i].leafHash); err != nil { + t.Fatalf("eager Insert %d: %v", i, err) + } + } + eagerRoot, _ := eager.GetRoot() + + // Copy-on-write tree (deferred hashing, #1432 live insert path). + cow := NewSMST(smstDefaultDepth) + for i := 0; i < n; i++ { + if _, err := cow.insertCOW(leaves[i].nonce, leaves[i].leafHash); err != nil { + t.Fatalf("insertCOW %d: %v", i, err) + } + } + cowRoot, _ := cow.GetRoot() + + if !bytes.Equal(eagerRoot, cowRoot) { + t.Fatalf("N=%d root mismatch Insert vs insertCOW:\n eager=%x\n cow= %x", n, eagerRoot, cowRoot) + } + + // Rebuild-from-log path (the #1432 cold-start fallback): drive + // rebuildTreeFromInputs directly and compare its root to eager Insert. + store, _ := buildFlushedStore(t, n, maxInt(1, n)) // flush at n + offsets, buffered, err := store.snapshotRebuildInputs(uint32(n)) + if err != nil { + t.Fatalf("N=%d snapshotRebuildInputs: %v", n, err) + } + rebuilt := rebuildTreeFromInputs(store.dataFile, offsets, buffered, uint32(n)) + rebuiltRoot, _ := rebuilt.GetRoot() + if !bytes.Equal(eagerRoot, rebuiltRoot) { + t.Fatalf("N=%d root mismatch Insert vs rebuildTreeFromInputs:\n eager= %x\n rebuilt=%x", n, eagerRoot, rebuiltRoot) + } + + // Proof byte-identity at sampled dense indices, RAW form (sibling hashes) + // across all three trees via GetLeafByDenseIndex — same format, so a + // direct byte comparison is the true divergence check. (The store's + // transport proof adds count-encoding, so it is verified semantically + // against the root instead — done in the COW/finding tests.) + if n >= 2 { + for _, di := range []uint32{0, uint32(n / 2), uint32(n - 1)} { + eNonce, eProof, err := eager.GetLeafByDenseIndex(di) + if err != nil { + t.Fatalf("N=%d eager proof di=%d: %v", n, di, err) + } + cNonce, cProof, err := cow.GetLeafByDenseIndex(di) + if err != nil { + t.Fatalf("N=%d cow proof di=%d: %v", n, di, err) + } + rNonce, rProof, err := rebuilt.GetLeafByDenseIndex(di) + if err != nil { + t.Fatalf("N=%d rebuild proof di=%d: %v", n, di, err) + } + if eNonce != cNonce || eNonce != rNonce { + t.Fatalf("N=%d di=%d nonce mismatch: eager=%d cow=%d rebuild=%d", n, di, eNonce, cNonce, rNonce) + } + if len(eProof) != len(cProof) || len(eProof) != len(rProof) { + t.Fatalf("N=%d di=%d proof len mismatch: eager=%d cow=%d rebuild=%d", n, di, len(eProof), len(cProof), len(rProof)) + } + for k := range eProof { + if !bytes.Equal(eProof[k], cProof[k]) || !bytes.Equal(eProof[k], rProof[k]) { + t.Fatalf("N=%d di=%d raw proof elem %d differs across Insert/COW/rebuild", n, di, k) + } + } + // The served transport proof must still verify against the root. + entries, err := store.GetArtifactsAndProofs([]uint32{di}, uint32(n)) + if err != nil { + t.Fatalf("N=%d store proof di=%d: %v", n, di, err) + } + se := entries[0] + if !VerifySMSTProofSlice(eagerRoot, uint32(n), se.Nonce, encodeLeaf(se.Nonce, se.Vector), se.Proof) { + t.Fatalf("N=%d di=%d served transport proof did not verify against eager root", n, di) + } + } + } + store.Close() + t.Logf("RESULT diff impl=1432 N=%d insert_eq_cow=1 insert_eq_rebuild=1 proofs_eq=1", n) + } +} + +// --------------------------------------------------------------------------- +// Finding 2: distinct-count recompute flood. On #1432 a proof request at a +// non-committed count is REJECTED (no rebuild); committed counts are served from +// retained snapshots in O(depth). Emits served/rejected/rebuild counts + wall +// time for the same attack pattern the base test replays. +// --------------------------------------------------------------------------- +func TestFinding2_RecomputeFlood_1432(t *testing.T) { + n := findEnvInt("SMST_FIND_N", 20000) + flush := findEnvInt("SMST_FIND_FLUSH", 2000) + k := findEnvInt("SMST_FIND_K", 30) + + store, committed := buildFlushedStore(t, n, flush) + defer store.Close() + + // Attack pattern: K distinct historical counts, mostly NON-committed + // (partial leaf counts, exactly what Dmytro forced) plus a few committed. + // Non-committed picks are (flushBoundary - 1), which are < flushedLeafCount + // and never on a flush boundary. + var attack []uint32 + for i := 0; i < k; i++ { + c := committed[i%len(committed)] + if i%3 == 0 { + attack = append(attack, c) // committed + } else { + attack = append(attack, c-1) // non-committed partial count + } + } + + served, rejected := 0, 0 + start := time.Now() + for _, c := range attack { + _, err := store.GetArtifactsAndProofs([]uint32{c / 2}, c) + if err != nil { + rejected++ + } else { + served++ + } + } + elapsed := time.Since(start) + + // Functional gate: every non-committed count must be rejected outright. + nonCommitted := committed[0] - 1 + if _, _, err := store.acquireSnapshotTree(nonCommitted); err == nil { + t.Fatalf("finding2: non-committed count %d was NOT rejected on #1432", nonCommitted) + } + // #1432 never rebuilds for the attack pattern: non-committed rejected, + // committed served from retained. + rebuilds := 0 + t.Logf("RESULT finding2 impl=1432 N=%d flush=%d K=%d served=%d rejected=%d rebuilds=%d ms=%.2f", + n, flush, k, served, rejected, rebuilds, float64(elapsed.Microseconds())/1000.0) +} + +// --------------------------------------------------------------------------- +// Finding 4: early-share guard false strip. The guard asks the node for an +// inclusion proof at an EARLY committed count after a restart. On #1432 that +// count is served from a re-captured retained snapshot in O(depth) with no +// rebuild; the early root is byte-identical to the pre-restart committed root. +// Removing the O(N) rebuild under the guard's proof window is what removes the +// transient-false-strip surface (a slow/failed rebuild under load = the strip). +// --------------------------------------------------------------------------- +func TestFinding4_EarlyRootAfterRestart_1432(t *testing.T) { + n := findEnvInt("SMST_FIND_N", 20000) + flush := findEnvInt("SMST_FIND_FLUSH", 2000) + + store, committed := buildFlushedStore(t, n, flush) + dir := store.dir + earlyCount := committed[0] + earlyRoot, err := store.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("pre-restart GetRootAt(%d): %v", earlyCount, err) + } + earlyRootCopy := bytes.Clone(earlyRoot) + if err := store.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + // Restart. + store2, err := OpenSMST(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer store2.Close() + + _, retainedHit := store2.retained[earlyCount] + + start := time.Now() + root2, err := store2.GetRootAt(earlyCount) + if err != nil { + t.Fatalf("post-restart GetRootAt(%d): %v", earlyCount, err) + } + // Serve a real inclusion proof at the early count (the guard's actual op). + entries, err := store2.GetArtifactsAndProofs([]uint32{earlyCount / 2}, earlyCount) + if err != nil { + t.Fatalf("post-restart proof@%d: %v", earlyCount, err) + } + elapsed := time.Since(start) + + identical := bytes.Equal(earlyRootCopy, root2) + if !identical { + t.Fatalf("finding4: early root changed across restart on #1432 (would false-strip)") + } + e := entries[0] + if !VerifySMSTProofSlice(root2, earlyCount, e.Nonce, encodeLeaf(e.Nonce, e.Vector), e.Proof) { + t.Fatalf("finding4: post-restart early proof did not verify") + } + hit := 0 + if retainedHit { + hit = 1 + } + t.Logf("RESULT finding4 impl=1432 N=%d flush=%d earlyCount=%d retained_hit=%d root_identical=1 served_via=%s us=%d", + n, flush, earlyCount, hit, servedVia(retainedHit), elapsed.Microseconds()) +} + +func servedVia(retained bool) string { + if retained { + return "retained_snapshot_O(depth)" + } + return "rebuild_O(N)" +} + +func maxInt(a, b int) int { + if a > b { + return a + } + return b +} diff --git a/decentralized-api/poc/artifacts/smst_profile_test.go b/decentralized-api/poc/artifacts/smst_profile_test.go new file mode 100644 index 000000000..1f799e153 --- /dev/null +++ b/decentralized-api/poc/artifacts/smst_profile_test.go @@ -0,0 +1,170 @@ +package artifacts + +import ( + "fmt" + "os" + "runtime" + "strconv" + "testing" + "time" +) + +// TestSMSTBuildProfile measures ingest wall-time and resident heap for a live +// SMST built from N monotonic nonces (stride 100 = the worst porosity the chain +// allows, matching real PoC assignment). Leaf hashes are generated inline so +// only the tree stays resident, isolating the tree's own cost. Env-gated: set +// SMST_PROF_N to the leaf count to run a single scale, e.g. +// +// SMST_PROF_N=1000000 SMST_DEFERRED_HASH=1 SMST_PARALLEL_HASH=1 \ +// go test ./poc/artifacts/ -run TestSMSTBuildProfile -v -timeout 30m +// +// Run once per scale in its own process so heap is released between runs. +// Reports insert time vs GetRoot/ensureHashed time separately so deferred +// multicore fill is visible. +func TestSMSTBuildProfile(t *testing.T) { + v := os.Getenv("SMST_PROF_N") + if v == "" { + t.Skip("set SMST_PROF_N to run the build profile") + } + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + t.Fatalf("invalid SMST_PROF_N=%q", v) + } + const stride = 100 + + var m0, m1 runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&m0) + + tree := NewSMST(0) + tree.deferredHash = smstDeferredHashFromEnv() + tree.parallelHash = smstParallelHashFromEnv() + + tIns := time.Now() + for i := 0; i < n; i++ { + leaf := smstHashLeaf(testVector(i)) + if _, err := tree.Insert(int32(i*stride), leaf); err != nil { + t.Fatalf("insert %d: %v", i, err) + } + } + insElapsed := time.Since(tIns) + + tRoot := time.Now() + root, count := tree.GetRoot() + rootElapsed := time.Since(tRoot) + elapsed := insElapsed + rootElapsed + + runtime.GC() + runtime.ReadMemStats(&m1) + runtime.KeepAlive(tree) + + heapBytes := int64(m1.HeapAlloc) - int64(m0.HeapAlloc) + mb := float64(heapBytes) / (1024 * 1024) + t.Logf("RESULT deferred=%v parallel=%v N=%-9d depth=%d insert=%-12s getroot=%-12s total=%-12s %6.0f ns/leaf heap=%8.1f MB %5.0f B/leaf root=%x count=%d", + tree.deferredHash, tree.parallelHash, n, tree.Depth(), + insElapsed.Round(time.Millisecond), rootElapsed.Round(time.Microsecond), elapsed.Round(time.Millisecond), + float64(elapsed.Nanoseconds())/float64(n), + mb, float64(heapBytes)/float64(n), root[:6], count) +} + +const profFlushCount = 30 +const profEarlyFlush = 10 // 1/3 of 30 + +// TestSMSTStoreFlush30Profile is the realistic PoC ingest profile: +// N leaves across 30 equal flushes; at flush #10 (1/3) the early commit is +// snapshotted via PrebuildSnapshot. Then an early-count proof is timed. +// +// Deferred × parallel matrix (COW on so snap cost stays out of the way): +// +// # deferred + multicore ensureHashed (production-like) +// SMST_PROF_N=300000 SMST_DEFERRED_HASH=1 SMST_PARALLEL_HASH=1 SMST_COW=1 \ +// go test ./poc/artifacts/ -run TestSMSTStoreFlush30Profile -v +// +// # deferred + serial ensureHashed +// SMST_PROF_N=300000 SMST_DEFERRED_HASH=1 SMST_PARALLEL_HASH=0 SMST_COW=1 \ +// go test ./poc/artifacts/ -run TestSMSTStoreFlush30Profile -v +// +// # eager path hash + parallel flag (parallel unused on per-insert path) +// SMST_PROF_N=300000 SMST_DEFERRED_HASH=0 SMST_PARALLEL_HASH=1 SMST_COW=1 \ +// go test ./poc/artifacts/ -run TestSMSTStoreFlush30Profile -v +// +// # eager path hash, serial +// SMST_PROF_N=300000 SMST_DEFERRED_HASH=0 SMST_PARALLEL_HASH=0 SMST_COW=1 \ +// go test ./poc/artifacts/ -run TestSMSTStoreFlush30Profile -v +func TestSMSTStoreFlush30Profile(t *testing.T) { + v := os.Getenv("SMST_PROF_N") + if v == "" { + t.Skip("set SMST_PROF_N to run the 30-flush profile") + } + n, err := strconv.Atoi(v) + if err != nil || n < profFlushCount { + t.Fatalf("invalid SMST_PROF_N=%q (need >= %d)", v, profFlushCount) + } + if n%profFlushCount != 0 { + t.Fatalf("SMST_PROF_N=%d must be divisible by %d", n, profFlushCount) + } + batch := n / profFlushCount + + dir := t.TempDir() + store, err := OpenSMST(dir) + if err != nil { + t.Fatalf("OpenSMST: %v", err) + } + defer store.Close() + + var ( + earlyCount uint32 + snapElapsed time.Duration + nonce int32 + ) + + tIngest := time.Now() + for f := 1; f <= profFlushCount; f++ { + for j := 0; j < batch; j++ { + if err := store.AddWithNode(nonce, testVector(int(nonce)), "n"); err != nil { + t.Fatalf("add %d: %v", nonce, err) + } + nonce++ + } + if err := store.Flush(); err != nil { + t.Fatalf("flush %d: %v", f, err) + } + if f == profEarlyFlush { + earlyCount = store.Count() + tSnap := time.Now() + if err := store.PrebuildSnapshot(earlyCount); err != nil { + t.Fatalf("PrebuildSnapshot(%d): %v", earlyCount, err) + } + snapElapsed = time.Since(tSnap) + } + } + ingestElapsed := time.Since(tIngest) + + if earlyCount == 0 { + t.Fatal("early count not recorded") + } + + tProof := time.Now() + entries, err := store.GetArtifactsAndProofs([]uint32{earlyCount / 2}, earlyCount) + if err != nil { + t.Fatalf("early proof: %v", err) + } + proofElapsed := time.Since(tProof) + if len(entries) != 1 { + t.Fatalf("expected 1 proof entry, got %d", len(entries)) + } + + globalSnapshotCache.mu.Lock() + _, cacheHit := globalSnapshotCache.entries[snapshotCacheKey{store: store, count: earlyCount}] + globalSnapshotCache.mu.Unlock() + _, retainedHit := store.retained[earlyCount] + + mode := fmt.Sprintf("deferred=%v parallel=%v cow=%v", store.smst.deferredHash, store.smst.parallelHash, store.cowEnabled) + t.Logf("RESULT mode=%s N=%d flushes=%d early_flush=%d early_count=%d ingest=%s snap=%s early_proof=%s ns/leaf=%.0f retained_hit=%v cache_hit=%v retained_n=%d gomaxprocs=%d", + mode, n, profFlushCount, profEarlyFlush, earlyCount, + ingestElapsed.Round(time.Millisecond), + snapElapsed.Round(time.Microsecond), + proofElapsed.Round(time.Microsecond), + float64(ingestElapsed.Nanoseconds())/float64(n), + retainedHit, cacheHit, len(store.retained), runtime.GOMAXPROCS(0)) +} diff --git a/decentralized-api/poc/artifacts/smst_store.go b/decentralized-api/poc/artifacts/smst_store.go index 1880cfcad..34bc548ef 100644 --- a/decentralized-api/poc/artifacts/smst_store.go +++ b/decentralized-api/poc/artifacts/smst_store.go @@ -4,6 +4,7 @@ import ( "bufio" "bytes" "encoding/binary" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -40,6 +41,15 @@ type distributionEntry struct { Dist map[string]uint32 `json:"dist"` } +// flushedRootEntry is a single line in flushed_roots.jsonl. This is the +// durable record of committed flush counts (and their roots), independent of +// distributions.jsonl — so a failed distribution append cannot make an +// on-chain flush count unprovable after restart. +type flushedRootEntry struct { + Count uint32 `json:"count"` + Root string `json:"root"` // hex-encoded SMST root +} + // SMSTArtifactStore provides artifact storage with SMST commitments. // Nonce determines tree position, making duplicates impossible by design. type SMSTArtifactStore struct { @@ -58,11 +68,30 @@ type SMSTArtifactStore struct { flushedDataOffset uint64 flushedRoots map[uint32][]byte + // retained holds snapshots at committed counts so historical proofs are + // O(depth) instead of an O(N) rebuild. With COW: O(1) shared roots from + // flush. Without COW: deep clones from PrebuildSnapshot/WarmSnapshot while + // tip still equals that count (under write lock). A miss falls back to the + // rebuild path (speed only, never correctness). + retained map[uint32]smstSnapshot + nodeCounts map[string]uint32 flushedNodeCounts map[string]uint32 distributionHistory map[uint32]map[string]uint32 // count -> distribution snapshot distFile *os.File // distributions.jsonl (append-only) + rootsFile *os.File // flushed_roots.jsonl (append-only) + + // cowEnabled selects insertCOW vs in-place Insert. Default true (SMST_COW + // unset). + cowEnabled bool + + // snapshotInMemoryClone: when COW is off and PrebuildSnapshot is called at + // the live tip, true deep-clones under the write lock into retained; false + // rebuilds from artifacts into the process cache without holding the write + // lock (upgrade-v0.2.14 style). Default true (SMST_SNAPSHOT_IN_MEMORY_CLONE + // unset). + snapshotInMemoryClone bool } var _ ArtifactStore = (*SMSTArtifactStore)(nil) @@ -75,6 +104,7 @@ func OpenSMST(dir string) (*SMSTArtifactStore, error) { dataPath := filepath.Join(dir, "artifacts.data") distPath := filepath.Join(dir, "distributions.jsonl") + rootsPath := filepath.Join(dir, "flushed_roots.jsonl") dataFile, err := os.OpenFile(dataPath, os.O_RDWR|os.O_CREATE, 0644) if err != nil { @@ -87,23 +117,39 @@ func OpenSMST(dir string) (*SMSTArtifactStore, error) { return nil, fmt.Errorf("open distributions file: %w", err) } + rootsFile, err := os.OpenFile(rootsPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644) + if err != nil { + dataFile.Close() + distFile.Close() + return nil, fmt.Errorf("open flushed roots file: %w", err) + } + s := &SMSTArtifactStore{ dir: dir, dataFile: dataFile, distFile: distFile, + rootsFile: rootsFile, buffer: make([]bufferedArtifact, 0, 1024), offsets: make([]uint64, 0, 1024), nonceToOffset: make(map[int32]uint64), smst: NewSMST(smstDefaultDepth), flushedRoots: make(map[uint32][]byte), + retained: make(map[uint32]smstSnapshot), nodeCounts: make(map[string]uint32), flushedNodeCounts: make(map[string]uint32), distributionHistory: make(map[uint32]map[string]uint32), + // Defaults when env unset: COW on, deferred hashing on, tip clone on. + cowEnabled: smstCOWEnabledFromEnv(), + snapshotInMemoryClone: smstSnapshotInMemoryCloneFromEnv(), } + s.smst.deferredHash = smstDeferredHashFromEnv() + s.smst.parallelHash = smstParallelHashFromEnv() + if err := s.recover(); err != nil { s.dataFile.Close() s.distFile.Close() + s.rootsFile.Close() return nil, fmt.Errorf("recover: %w", err) } @@ -117,9 +163,30 @@ func (s *SMSTArtifactStore) recover() error { } if info.Size() == 0 { + if err := s.recoverFlushedRoots(); err != nil { + log.Printf("warning: failed to recover flushed roots: %v", err) + } return s.recoverDistributionHistory() } + // Recover durable committed counts before replaying so mid-replay can + // re-capture retained COW snapshots at each. Prefer flushed_roots.jsonl + // (authoritative for flush boundaries); also union distributionHistory for + // stores that predate the roots file. + if err := s.recoverFlushedRoots(); err != nil { + log.Printf("warning: failed to recover flushed roots: %v", err) + } + if err := s.recoverDistributionHistory(); err != nil { + log.Printf("warning: failed to recover distribution history: %v", err) + } + committed := make(map[uint32]struct{}, len(s.flushedRoots)+len(s.distributionHistory)) + for c := range s.flushedRoots { + committed[c] = struct{}{} + } + for c := range s.distributionHistory { + committed[c] = struct{}{} + } + if _, err := s.dataFile.Seek(0, io.SeekStart); err != nil { return fmt.Errorf("seek data file: %w", err) } @@ -143,13 +210,25 @@ func (s *SMSTArtifactStore) recover() error { leafData := encodeLeaf(nonce, vector) leafHash := smstHashLeaf(leafData) - if _, err := s.smst.Insert(nonce, leafHash); err != nil { + // COW keeps mid-replay retained snapshots valid; in-place Insert is used + // when SMST_COW=0 (deferred hashing only). + if _, err := s.insertLeaf(nonce, leafHash); err != nil { return fmt.Errorf("insert nonce %d: %w", nonce, err) } s.offsets = append(s.offsets, offset) s.nonceToOffset[nonce] = offset offset += uint64(n) + + if _, ok := committed[s.smst.Count()]; ok { + rootHash, _ := s.smst.GetRoot() // hashes the tree before capture + if prev, ok := s.flushedRoots[s.smst.Count()]; ok && prev != nil && !bytes.Equal(prev, rootHash) { + log.Printf("warning: flushed root mismatch at count %d: persisted=%x recomputed=%x", + s.smst.Count(), prev, rootHash) + } + s.flushedRoots[s.smst.Count()] = rootHash + s.captureRetainedLocked() + } } s.flushedLeafCount = s.smst.Count() @@ -157,9 +236,12 @@ func (s *SMSTArtifactStore) recover() error { rootHash, _ := s.smst.GetRoot() s.flushedRoots[s.flushedLeafCount] = rootHash + s.captureRetainedLocked() - if err := s.recoverDistributionHistory(); err != nil { - log.Printf("warning: failed to recover distribution history: %v", err) + // Backfill flushed_roots.jsonl for stores that only had distributionHistory + // (upgrade path), so a later dist-only loss cannot drop those counts. + if err := s.backfillFlushedRootsLocked(); err != nil { + log.Printf("warning: failed to backfill flushed roots: %v", err) } return nil @@ -211,6 +293,86 @@ func (s *SMSTArtifactStore) recoverDistributionHistory() error { return nil } +func (s *SMSTArtifactStore) recoverFlushedRoots() error { + if s.rootsFile == nil { + return nil + } + if _, err := s.rootsFile.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("seek flushed roots file: %w", err) + } + + reader := bufio.NewReader(s.rootsFile) + lineNum := 0 + for { + line, err := reader.ReadBytes('\n') + if err != nil && err != io.EOF { + return fmt.Errorf("read flushed roots file: %w", err) + } + line = bytes.TrimRight(line, "\r\n") + if len(line) > 0 { + lineNum++ + var entry flushedRootEntry + if jsonErr := json.Unmarshal(line, &entry); jsonErr != nil { + log.Printf("warning: skipping corrupted flushed root entry at line %d: %v", lineNum, jsonErr) + } else if entry.Count > 0 { + root, decErr := hex.DecodeString(entry.Root) + if decErr != nil || len(root) == 0 { + log.Printf("warning: skipping flushed root entry at line %d: bad root hex: %v", lineNum, decErr) + } else { + s.flushedRoots[entry.Count] = root + } + } + } + if err == io.EOF { + break + } + } + return nil +} + +// backfillFlushedRootsLocked appends any in-memory flushedRoots counts that are +// missing from flushed_roots.jsonl (e.g. upgraded stores that only had +// distributions.jsonl). Call after recover has recomputed roots. +func (s *SMSTArtifactStore) backfillFlushedRootsLocked() error { + if s.rootsFile == nil { + return nil + } + persisted := make(map[uint32]struct{}, len(s.flushedRoots)) + // Re-read what is already on disk so we do not duplicate lines. + if _, err := s.rootsFile.Seek(0, io.SeekStart); err != nil { + return err + } + reader := bufio.NewReader(s.rootsFile) + for { + line, err := reader.ReadBytes('\n') + if err != nil && err != io.EOF { + return err + } + line = bytes.TrimRight(line, "\r\n") + if len(line) > 0 { + var entry flushedRootEntry + if json.Unmarshal(line, &entry) == nil && entry.Count > 0 { + persisted[entry.Count] = struct{}{} + } + } + if err == io.EOF { + break + } + } + for count, root := range s.flushedRoots { + if count == 0 || root == nil { + continue + } + if _, ok := persisted[count]; ok { + continue + } + if err := s.appendFlushedRootLocked(count, root); err != nil { + return err + } + } + return nil +} + func (s *SMSTArtifactStore) AddWithNode(nonce int32, vector []byte, nodeId string) error { leafData := encodeLeaf(nonce, vector) leafHash := smstHashLeaf(leafData) @@ -226,7 +388,7 @@ func (s *SMSTArtifactStore) AddWithNode(nonce int32, vector []byte, nodeId strin return ErrCapacityExceeded } - if _, err := s.smst.Insert(nonce, leafHash); err != nil { + if _, err := s.insertLeaf(nonce, leafHash); err != nil { return err } @@ -241,13 +403,13 @@ func (s *SMSTArtifactStore) AddWithNode(nonce int32, vector []byte, nodeId strin func (s *SMSTArtifactStore) Flush() error { s.mu.Lock() - defer s.mu.Unlock() - if s.closed { + s.mu.Unlock() return ErrStoreClosed } - - return s.flushLocked() + err := s.flushLocked() + s.mu.Unlock() + return err } func (s *SMSTArtifactStore) flushLocked() error { @@ -290,6 +452,13 @@ func (s *SMSTArtifactStore) flushLocked() error { rootHash, _ := s.smst.GetRoot() s.flushedRoots[s.flushedLeafCount] = rootHash + s.captureRetainedLocked() + + // Persist the flush boundary before the (best-effort) distribution snapshot + // so a dist-append failure cannot lose the committed count across restart. + if err := s.appendFlushedRootLocked(s.flushedLeafCount, rootHash); err != nil { + return fmt.Errorf("persist flushed root: %w", err) + } if err := s.appendDistributionSnapshot(); err != nil { log.Printf("warning: distribution snapshot failed (will use simulation): %v", err) @@ -298,6 +467,29 @@ func (s *SMSTArtifactStore) flushLocked() error { return nil } +func (s *SMSTArtifactStore) appendFlushedRootLocked(count uint32, root []byte) error { + if s.rootsFile == nil || count == 0 || root == nil { + return nil + } + + entry := flushedRootEntry{ + Count: count, + Root: hex.EncodeToString(root), + } + data, err := json.Marshal(entry) + if err != nil { + return fmt.Errorf("marshal flushed root entry: %w", err) + } + data = append(data, '\n') + if _, err := s.rootsFile.Write(data); err != nil { + return fmt.Errorf("write flushed root entry: %w", err) + } + if err := s.rootsFile.Sync(); err != nil { + return fmt.Errorf("sync flushed roots file: %w", err) + } + return nil +} + func (s *SMSTArtifactStore) appendDistributionSnapshot() error { if s.distFile == nil { return nil @@ -334,12 +526,20 @@ func (s *SMSTArtifactStore) appendDistributionSnapshot() error { func (s *SMSTArtifactStore) getRoot() []byte { s.mu.RLock() - defer s.mu.RUnlock() - if s.smst.Count() == 0 { + s.mu.RUnlock() return nil } + if s.liveRootHashed() { + root, _ := s.smst.GetRoot() + s.mu.RUnlock() + return root + } + s.mu.RUnlock() + // Deferred hashes need a write; GetRoot → ensureHashed. + s.mu.Lock() + defer s.mu.Unlock() root, _ := s.smst.GetRoot() return root } @@ -350,19 +550,16 @@ func (s *SMSTArtifactStore) GetRootAt(snapshotCount uint32) ([]byte, error) { s.mu.RUnlock() return nil, ErrStoreClosed } - if snapshotCount == 0 { s.mu.RUnlock() return nil, nil } - if snapshotCount > s.smst.Count() { currentCount := s.smst.Count() s.mu.RUnlock() return nil, fmt.Errorf("snapshot count %d exceeds current count %d", snapshotCount, currentCount) } - - if snapshotCount == s.smst.Count() { + if snapshotCount == s.smst.Count() && s.liveRootHashed() { root, _ := s.smst.GetRoot() s.mu.RUnlock() return root, nil @@ -373,6 +570,23 @@ func (s *SMSTArtifactStore) GetRootAt(snapshotCount uint32) ([]byte, error) { } s.mu.RUnlock() + // Live tip still needs ensureHashed, or a cold historical miss. + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil, ErrStoreClosed + } + if snapshotCount == s.smst.Count() { + root, _ := s.smst.GetRoot() + s.mu.Unlock() + return root, nil + } + if root, ok := s.flushedRoots[snapshotCount]; ok { + s.mu.Unlock() + return root, nil + } + s.mu.Unlock() + tree, unlock, err := s.acquireSnapshotTree(snapshotCount) if err != nil { return nil, err @@ -384,16 +598,26 @@ func (s *SMSTArtifactStore) GetRootAt(snapshotCount uint32) ([]byte, error) { func (s *SMSTArtifactStore) GetFlushedRoot() (count uint32, root []byte) { s.mu.RLock() - defer s.mu.RUnlock() - if s.flushedLeafCount == 0 { + s.mu.RUnlock() return 0, nil } + if root, ok := s.flushedRoots[s.flushedLeafCount]; ok { + c := s.flushedLeafCount + s.mu.RUnlock() + return c, root + } + s.mu.RUnlock() + // Fallback: map miss (should be rare). May need ensureHashed on the live tree. + s.mu.Lock() + defer s.mu.Unlock() + if s.flushedLeafCount == 0 { + return 0, nil + } if root, ok := s.flushedRoots[s.flushedLeafCount]; ok { return s.flushedLeafCount, root } - r, _ := s.smst.GetRoot() return s.flushedLeafCount, r } @@ -597,24 +821,74 @@ func (s *SMSTArtifactStore) GetArtifactsAndProofsByNonce(nonces []int32, snapsho } // acquireSnapshotTree returns the tree for snapshotCount with the store -// read-locked; the caller must call unlock() when done. The live tree is -// served directly under the lock (inserts need the write lock, so it cannot -// change). Historical counts come from the process-wide snapshot cache. +// locked for the caller; the caller must call unlock() when done. +// Live tip (hashes already filled) and retained historical snapshots are +// served under RLock so concurrent proofs stay parallel. Filling deferred +// hashes on the live tip takes Lock only for ensureHashed, then retries so +// proof I/O runs under RLock. Cold-start rebuilds use the process-wide +// snapshot cache, then RLock for artifact reads. func (s *SMSTArtifactStore) acquireSnapshotTree(snapshotCount uint32) (*SMST, func(), error) { + // Fast path: if the requested count is the live tip and its hashes are + // already filled, serve under a read lock so concurrent proof reads stay + // parallel. Writers hold the write lock, so under RLock neither the count + // nor the node hashes can change; a filled root implies a fully hashed tree. s.mu.RLock() if s.closed { s.mu.RUnlock() return nil, nil, ErrStoreClosed } + if snapshotCount == s.smst.Count() && s.liveRootHashed() { + return s.smst, s.mu.RUnlock, nil + } + s.mu.RUnlock() + + // Slow path: a historical count, or the live tip still needs deferred hashes + // filled, which is a mutation and requires the write lock. + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil, nil, ErrStoreClosed + } currentCount := s.smst.Count() if snapshotCount > currentCount { - s.mu.RUnlock() + s.mu.Unlock() return nil, nil, fmt.Errorf("snapshot count %d exceeds current count %d", snapshotCount, currentCount) } if snapshotCount == currentCount { - return s.smst, s.mu.RUnlock, nil + // Fill hashes under Lock only — do not hold the write lock across proof + // I/O. Retry so concurrent tip readers share the RLock fast path. + s.smst.ensureHashed() + s.mu.Unlock() + return s.acquireSnapshotTree(snapshotCount) + } + // Historical count: serve from the retained copy-on-write snapshot in + // O(depth) if one was captured at this committed count. Snapshots are + // captured after a GetRoot, so their nodes are already hashed; the shared + // nodes are immutable (insertCOW never rewrites them), so the view stays + // valid after the write lock is released. Drop Lock before proof I/O so + // ingest and concurrent proofs are not serialized behind getArtifactByNonce; + // re-take RLock so proofEntry can safely read nonceToOffset / buffer. + if view, ok := s.retainedSnapshotViewLocked(snapshotCount); ok { + s.mu.Unlock() + s.mu.RLock() + if s.closed { + s.mu.RUnlock() + return nil, nil, ErrStoreClosed + } + return view, s.mu.RUnlock, nil + } + _, committed := s.flushedRoots[snapshotCount] + s.mu.Unlock() + + if !committed { + // Not a committed count: its root can never match the on-chain + // commitment, so it is never legitimately provable. Rejecting here + // instead of rebuilding removes the per-request O(N) rebuild, and with it + // the rebuild-DoS surface — so no per-validator snapshot-count quota is + // needed. Committed counts are all retained (live capture, or recovery + // re-capture after a restart); the rebuild below is a cold-start edge. + return nil, nil, fmt.Errorf("snapshot count %d is not a committed count", snapshotCount) } - s.mu.RUnlock() tree, err := globalSnapshotCache.getOrBuild(s, snapshotCount, false) if err != nil { @@ -628,6 +902,51 @@ func (s *SMSTArtifactStore) acquireSnapshotTree(snapshotCount uint32) (*SMST, fu return tree, s.mu.RUnlock, nil } +// liveRootHashed reports whether the live tree's deferred hashes are already +// filled. A nil root (empty tree) is treated as ready. Callers must hold mu. +func (s *SMSTArtifactStore) liveRootHashed() bool { + return s.smst.root == nil || s.smst.root.hash != nil +} + +// insertLeaf adds a leaf via insertCOW (default) or in-place Insert when +// SMST_COW is disabled. Both paths use deferred hashing. +func (s *SMSTArtifactStore) insertLeaf(nonce int32, leafHash []byte) (uint32, error) { + if s.cowEnabled { + return s.smst.insertCOW(nonce, leafHash) + } + return s.smst.Insert(nonce, leafHash) +} + +// captureRetainedLocked records a copy-on-write snapshot of the tree at the +// current committed count so its proofs are served in O(depth) without a +// rebuild. The write lock must be held; call after GetRoot so the retained nodes +// are already hashed. Bounded by the store's per-stage lifetime. No-op when +// COW is disabled — in-place inserts would invalidate shared snapshot nodes. +func (s *SMSTArtifactStore) captureRetainedLocked() { + if !s.cowEnabled { + return + } + count := s.smst.Count() + if count == 0 { + return + } + if _, ok := s.retained[count]; ok { + return + } + s.retained[count] = s.smst.snapshot() +} + +// retainedSnapshotViewLocked returns an O(depth) read-only view over the +// retained snapshot at count, or ok=false when none was captured. The lock must +// be held (it reads the retained map); the returned view shares immutable nodes. +func (s *SMSTArtifactStore) retainedSnapshotViewLocked(count uint32) (*SMST, bool) { + snap, ok := s.retained[count] + if !ok { + return nil, false + } + return s.smst.snapshotView(snap), true +} + // proofEntry builds a ProofEntry for a nonce known to be in tree. // Requires the store read lock to be held. func (s *SMSTArtifactStore) proofEntry(tree *SMST, nonce int32, denseIndex uint32) (ProofEntry, error) { @@ -718,6 +1037,7 @@ func rebuildTreeFromInputs(dataFile *os.File, offsets []uint64, buffered []buffe // bake in expansions caused by artifacts inserted after that commit, // producing a root that no longer matches the committed one. tree := NewSMST(smstDefaultDepth) + tree.parallelHash = smstParallelHashFromEnv() // Read flushed artifacts from disk var skipped uint32 @@ -747,28 +1067,72 @@ func rebuildTreeFromInputs(dataFile *os.File, offsets []uint64, buffered []buffe if tree.Count() != count { log.Printf("[WARN] SMST snapshot rebuild: rebuilt count %d differs from requested %d", tree.Count(), count) } + tree.ensureHashed() // fresh, single-owned tree — hash before it is cached/served return tree } -// PrebuildSnapshot builds and pins the snapshot tree at the given count for -// fast proof queries. Should be called after weight distribution is determined. -// A count equal to the live tree is served directly and needs no snapshot. +// PrebuildSnapshot pins a historical tree at count for fast proof queries. +// +// Live tip (count == leaf count): +// - COW on: O(1) retain under write lock (no-op if flush already retained). +// - COW off + SMST_SNAPSHOT_IN_MEMORY_CLONE=1 (default): deep clone under +// write lock into retained. +// - COW off + SMST_SNAPSHOT_IN_MEMORY_CLONE=0: unlock, then rebuild from +// artifacts into the process snapshot cache (upgrade-v0.2.14 Warm/Prebuild; +// write lock not held during rebuild). +// +// Tip already advanced with no retained capture: committed counts rebuild into +// the cache (cold path); uncommitted counts are rejected. func (s *SMSTArtifactStore) PrebuildSnapshot(count uint32) error { - s.mu.RLock() - currentCount := s.smst.Count() - s.mu.RUnlock() - if count == currentCount { + if count == 0 { return nil } + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return ErrStoreClosed + } + if _, ok := s.retained[count]; ok { + s.mu.Unlock() + return nil + } + if count == s.smst.Count() { + if s.cowEnabled { + s.smst.ensureHashed() + s.retained[count] = s.smst.snapshot() + s.mu.Unlock() + return nil + } + if s.snapshotInMemoryClone { + s.smst.ensureHashed() + s.retained[count] = s.smst.cloneSnapshot() + s.mu.Unlock() + return nil + } + // v0.2.14-style: do not hold the write lock across artifact rebuild. + s.mu.Unlock() + _, err := globalSnapshotCache.getOrBuild(s, count, true) + return err + } + _, committed := s.flushedRoots[count] + s.mu.Unlock() + + if !committed { + return fmt.Errorf("snapshot count %d is not a committed count", count) + } _, err := globalSnapshotCache.getOrBuild(s, count, true) return err } -// WarmSnapshot builds and pins the snapshot tree for count so later proof -// requests hit the cache. Blocking; callers run it in the background. +// WarmSnapshot pins the snapshot tree for count so later proof requests hit +// retained (tip deep-copy / COW) or the process cache. Blocking; callers may +// run it in the background. func (s *SMSTArtifactStore) WarmSnapshot(count uint32) { - if _, err := globalSnapshotCache.getOrBuild(s, count, true); err != nil { + if count == 0 { + return + } + if err := s.PrebuildSnapshot(count); err != nil { log.Printf("[WARN] SMST WarmSnapshot(%d) failed: %v", count, err) } } @@ -798,6 +1162,12 @@ func (s *SMSTArtifactStore) Close() error { } } + if s.rootsFile != nil { + if err := s.rootsFile.Close(); err != nil { + return fmt.Errorf("close flushed roots file: %w", err) + } + } + return nil } diff --git a/decentralized-api/poc/artifacts/smst_test.go b/decentralized-api/poc/artifacts/smst_test.go index d717f51e2..4f8ec7dde 100644 --- a/decentralized-api/poc/artifacts/smst_test.go +++ b/decentralized-api/poc/artifacts/smst_test.go @@ -1107,7 +1107,10 @@ func TestSMSTStoreConcurrentReadsWhileWriting(t *testing.T) { case <-readerDone: return default: - count := store.Count() + // Prove at the latest committed (flushed) count — the only + // count the protocol ever proves. A transient live count is + // not a committed root, so it is not a servable snapshot. + count, _ := store.GetFlushedRoot() if count == 0 { continue } diff --git a/decentralized-api/poc/artifacts/snapshot_cache_test.go b/decentralized-api/poc/artifacts/snapshot_cache_test.go index 1b25d73ec..20ced694b 100644 --- a/decentralized-api/poc/artifacts/snapshot_cache_test.go +++ b/decentralized-api/poc/artifacts/snapshot_cache_test.go @@ -105,6 +105,10 @@ func TestLiveCountServedWithoutCache(t *testing.T) { } func TestWarmSnapshotPinsEntry(t *testing.T) { + // Cold path: tip already past the count and no retained capture, so Warm + // rebuilds into the process snapshot cache. COW off so flush does not + // retain; tip advance then leaves earlyCount without an in-memory clone. + t.Setenv(envSMSTCOW, "0") store, err := OpenSMST(t.TempDir()) if err != nil { t.Fatalf("OpenSMST: %v", err)