|
| 1 | +// Copyright 2025 Democratized Data Foundation |
| 2 | +// |
| 3 | +// Use of this software is governed by the Business Source License |
| 4 | +// included in the file licenses/BSL.txt. |
| 5 | +// |
| 6 | +// As of the Change Date specified in that file, in accordance with |
| 7 | +// the Business Source License, use of this software will be governed |
| 8 | +// by the Apache License, Version 2.0, included in the file |
| 9 | +// licenses/APL.txt. |
| 10 | + |
| 11 | +package datastore |
| 12 | + |
| 13 | +import ( |
| 14 | + "context" |
| 15 | + "errors" |
| 16 | + "time" |
| 17 | + |
| 18 | + "github.com/ipfs/go-cid" |
| 19 | + |
| 20 | + "github.com/sourcenetwork/corekv" |
| 21 | + "github.com/sourcenetwork/corekv/namespace" |
| 22 | + "github.com/sourcenetwork/defradb/internal/db/blockowner" |
| 23 | +) |
| 24 | + |
| 25 | +// orphanDeleteBatchSize is how many markers share one delete transaction. A merge |
| 26 | +// that touches any marker in a batch aborts the whole batch, so this is kept small |
| 27 | +// enough that redoing one on a later pass is cheap, while still amortising the commit |
| 28 | +// across many blocks. |
| 29 | +const orphanDeleteBatchSize = 256 |
| 30 | + |
| 31 | +// SweepResult reports what one call to ReclaimOrphanBlocks did. |
| 32 | +type SweepResult struct { |
| 33 | + // NextKey is where the next call resumes, nil once the index has been swept end to |
| 34 | + // end. It is only meaningful when the call returned no error. |
| 35 | + NextKey []byte |
| 36 | + Scanned int |
| 37 | + Reclaimed int |
| 38 | + // Repaired counts markers cleared from blocks a document still owns, which is a |
| 39 | + // marker that outlived the merge that claimed the block. |
| 40 | + Repaired int |
| 41 | + // Conflicts counts batches abandoned because a merge committed against a marker they |
| 42 | + // held. Their blocks stay marked and are reconsidered on the next full pass. |
| 43 | + Conflicts int |
| 44 | +} |
| 45 | + |
| 46 | +// ReclaimOrphanBlocks deletes blocks that were fetched during P2P sync but never merged |
| 47 | +// into a document, identified by a to-merge marker older than cutoff. History prune |
| 48 | +// never reaches these blocks because it walks out from documents, so without this sweep |
| 49 | +// they accumulate for the life of the store. |
| 50 | +// |
| 51 | +// It resumes from startKey, nil beginning at the start of the marker index, and scans at |
| 52 | +// most scanLimit markers per call. The scan only nominates candidates: it collects them |
| 53 | +// without mutating the iterator, then reclaimBatch re-decides each one under the |
| 54 | +// transaction that deletes it. |
| 55 | +// |
| 56 | +// Only markers carrying a timestamp are eligible. A marker without one was written before |
| 57 | +// the index recorded fetch times, by a store that also predates the block-owner edges this |
| 58 | +// sweep reads, so neither the age check nor the ownership check holds for it and its block |
| 59 | +// is left in place. |
| 60 | +func ReclaimOrphanBlocks( |
| 61 | + ctx context.Context, |
| 62 | + rootstore corekv.TxnReaderWriter, |
| 63 | + cutoff time.Time, |
| 64 | + startKey []byte, |
| 65 | + scanLimit int, |
| 66 | +) (SweepResult, error) { |
| 67 | + blockNS := namespace.Wrap(rootstore, []byte{blockStoreKey}) |
| 68 | + |
| 69 | + candidates, nextKey, scanned, err := collectExpiredMarkers(ctx, blockNS, cutoff, startKey, scanLimit) |
| 70 | + if err != nil { |
| 71 | + return SweepResult{Scanned: scanned}, err |
| 72 | + } |
| 73 | + result := SweepResult{NextKey: nextKey, Scanned: scanned} |
| 74 | + |
| 75 | + for start := 0; start < len(candidates); start += orphanDeleteBatchSize { |
| 76 | + // The loop commits one batch at a time and does not stop on its own, so |
| 77 | + // cancellation is checked here rather than only between sweeps. |
| 78 | + if err := ctx.Err(); err != nil { |
| 79 | + return result, err |
| 80 | + } |
| 81 | + end := min(start+orphanDeleteBatchSize, len(candidates)) |
| 82 | + if err := reclaimBatch(ctx, rootstore, cutoff, candidates[start:end], &result); err != nil { |
| 83 | + return result, err |
| 84 | + } |
| 85 | + } |
| 86 | + return result, nil |
| 87 | +} |
| 88 | + |
| 89 | +// reclaimBatch deletes one batch of orphaned blocks in a single transaction, re-reading |
| 90 | +// each marker inside it so the decision to delete and the delete itself are atomic, and |
| 91 | +// accumulates what it did into result. |
| 92 | +// |
| 93 | +// The re-read is what makes the sweep safe, and it rests on one invariant: a merge that |
| 94 | +// takes ownership of a block clears that block's to-merge marker in the same transaction. |
| 95 | +// Reading the marker here puts it in the transaction's read set, so that merge's write to |
| 96 | +// the same key makes this commit conflict and the batch is abandoned with nothing deleted. |
| 97 | +// The markers survive for the next full pass. |
| 98 | +func reclaimBatch( |
| 99 | + ctx context.Context, |
| 100 | + rootstore corekv.TxnReaderWriter, |
| 101 | + cutoff time.Time, |
| 102 | + markers [][]byte, |
| 103 | + result *SweepResult, |
| 104 | +) error { |
| 105 | + txn := rootstore.NewTxn(false) |
| 106 | + defer txn.Discard() |
| 107 | + |
| 108 | + // corekv takes the transaction from the context, so the stores below join it rather |
| 109 | + // than committing per call. |
| 110 | + txnCtx := corekv.SetCtxTxn(ctx, txn) |
| 111 | + blockNS := namespace.Wrap(rootstore, []byte{blockStoreKey}) |
| 112 | + systemNS := SystemstoreFrom(rootstore) |
| 113 | + |
| 114 | + reclaimed, repaired := 0, 0 |
| 115 | + for _, marker := range markers { |
| 116 | + value, err := blockNS.Get(txnCtx, marker) |
| 117 | + if errors.Is(err, corekv.ErrNotFound) { |
| 118 | + // A merge cleared the marker after the scan nominated it. |
| 119 | + continue |
| 120 | + } |
| 121 | + if err != nil { |
| 122 | + return err |
| 123 | + } |
| 124 | + if t, decoded := toMergeTime(value); !decoded || !t.Before(cutoff) { |
| 125 | + // Untimestamped, or re-written since the scan. |
| 126 | + continue |
| 127 | + } |
| 128 | + |
| 129 | + // A marker key is the to-merge prefix followed by the block's CID. |
| 130 | + blockCID, err := cid.Cast(marker[1:]) |
| 131 | + if err != nil { |
| 132 | + // Not a CID, so ownership cannot be checked. Leave it rather than guess. |
| 133 | + continue |
| 134 | + } |
| 135 | + |
| 136 | + owned, err := blockowner.HasOwners(txnCtx, systemNS, blockCID) |
| 137 | + if err != nil { |
| 138 | + return err |
| 139 | + } |
| 140 | + if owned { |
| 141 | + // A committed document owns the block, so the marker is stale rather than the |
| 142 | + // block being garbage. Ownership decides that, not the marker's presence. |
| 143 | + // Drop the marker, keep the block. |
| 144 | + if err := blockNS.Delete(txnCtx, marker); err != nil { |
| 145 | + return err |
| 146 | + } |
| 147 | + repaired++ |
| 148 | + continue |
| 149 | + } |
| 150 | + |
| 151 | + // Both deletes land in one commit, so a crash leaves the block and its marker |
| 152 | + // either both present, and reclaimable again on a later pass, or both gone. |
| 153 | + if err := blockNS.Delete(txnCtx, marker[1:]); err != nil { |
| 154 | + return err |
| 155 | + } |
| 156 | + if err := blockNS.Delete(txnCtx, marker); err != nil { |
| 157 | + return err |
| 158 | + } |
| 159 | + reclaimed++ |
| 160 | + } |
| 161 | + |
| 162 | + if err := txn.Commit(); err != nil { |
| 163 | + if errors.Is(err, corekv.ErrTxnConflict) { |
| 164 | + result.Conflicts++ |
| 165 | + return nil |
| 166 | + } |
| 167 | + return err |
| 168 | + } |
| 169 | + result.Reclaimed += reclaimed |
| 170 | + result.Repaired += repaired |
| 171 | + return nil |
| 172 | +} |
| 173 | + |
| 174 | +// collectExpiredMarkers scans the to-merge index from startKey and returns the keys of |
| 175 | +// markers that carry a timestamp older than cutoff, along with the key to resume from |
| 176 | +// next. These are candidates only; each is re-checked under a transaction before anything |
| 177 | +// is deleted. Kept keys are copied because the iterator reuses its buffers across Next. |
| 178 | +func collectExpiredMarkers( |
| 179 | + ctx context.Context, |
| 180 | + blockNS corekv.ReaderWriter, |
| 181 | + cutoff time.Time, |
| 182 | + startKey []byte, |
| 183 | + scanLimit int, |
| 184 | +) (expired [][]byte, nextKey []byte, scanned int, err error) { |
| 185 | + iter, err := blockNS.Iterator(ctx, corekv.IterOptions{Prefix: []byte{toMergeIndexPrefix}}) |
| 186 | + if err != nil { |
| 187 | + return nil, nil, 0, err |
| 188 | + } |
| 189 | + defer func() { |
| 190 | + if cerr := iter.Close(); cerr != nil && err == nil { |
| 191 | + expired, nextKey, err = nil, nil, cerr |
| 192 | + } |
| 193 | + }() |
| 194 | + |
| 195 | + ok, err := seekMarker(iter, startKey) |
| 196 | + for ok && err == nil { |
| 197 | + // The scan walks up to scanLimit markers and the iterator does not stop on its |
| 198 | + // own, so cancellation is checked here rather than only between sweeps. |
| 199 | + if cerr := ctx.Err(); cerr != nil { |
| 200 | + return nil, nil, scanned, cerr |
| 201 | + } |
| 202 | + if scanned == scanLimit { |
| 203 | + // The current marker is unexamined; resume from it next call. |
| 204 | + return expired, copyKey(iter.Key()), scanned, nil |
| 205 | + } |
| 206 | + var value []byte |
| 207 | + if value, err = iter.Value(); err != nil { |
| 208 | + break |
| 209 | + } |
| 210 | + if t, decoded := toMergeTime(value); decoded && t.Before(cutoff) { |
| 211 | + expired = append(expired, copyKey(iter.Key())) |
| 212 | + } |
| 213 | + scanned++ |
| 214 | + ok, err = iter.Next() |
| 215 | + } |
| 216 | + if err != nil { |
| 217 | + return nil, nil, scanned, err |
| 218 | + } |
| 219 | + return expired, nil, scanned, nil |
| 220 | +} |
| 221 | + |
| 222 | +// seekMarker positions the iterator at the first marker to examine: the beginning of |
| 223 | +// the index when startKey is nil, otherwise startKey, or the next marker if startKey |
| 224 | +// was deleted since the previous call. |
| 225 | +func seekMarker(iter corekv.Iterator, startKey []byte) (bool, error) { |
| 226 | + if startKey == nil { |
| 227 | + return iter.Next() |
| 228 | + } |
| 229 | + return iter.Seek(startKey) |
| 230 | +} |
| 231 | + |
| 232 | +func copyKey(k []byte) []byte { |
| 233 | + c := make([]byte, len(k)) |
| 234 | + copy(c, k) |
| 235 | + return c |
| 236 | +} |
0 commit comments