OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-note commits / 50d6345

Add push-based GC orphan queue to MediaStore
Adds scheduleOrphan/cancelOrphan/clearIfStillOrphan methods with configurable grace period and isReferencedAnywhere callback. Orphaned media hashes are cleared after the grace period unless re-referenced or cancelled. Includes 4 brittle tests covering clear, cancel, still-referenced, and idempotent scheduling. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
50d63459adc6810a85f5e6705e7a42ee17ee4ca2
by plankatron <fleeky@gmail.com> · 2026-03-19T16:26:16.000Z

Changed files (2) +182 -0

lib/media-store.js +68 -0
……@@ -13,6 +13,10 @@ export class MediaStore {
1313 keyEncoding: 'utf-8',
1414 valueEncoding: 'json'
1515 })
16
17 this._orphanQueue = new Map()
18 this._graceMs = opts.graceMs != null ? opts.graceMs : 7 * 24 * 60 * 60 * 1000
19 this.isReferencedAnywhere = opts.isReferencedAnywhere || (async () => false)
1620 }
1721
1822 async ready () {
……@@ -55,6 +59,70 @@ export class MediaStore {
5559 const entry = await this._index.get('blob:' + hash)
5660 return entry ? entry.value : null
5761 }
62
63 async clear (hash) {
64 const entry = await this._index.get('blob:' + hash)
65 if (!entry) return 0
66
67 const blobId = entry.value.blobId
68 const result = await this._blobs.clear(blobId, { diff: true })
69 await this._index.del('blob:' + hash)
70
71 return result ? result.blocks : 0
72 }
73
74 getStream (hash, range) {
75 const self = this
76 return (async function * () {
77 const entry = await self._index.get('blob:' + hash)
78 if (!entry) return
79
80 const blobId = entry.value.blobId
81 const opts = {}
82 if (range) {
83 if (range.start !== undefined) opts.start = range.start
84 if (range.end !== undefined) opts.end = range.end
85 else if (range.length !== undefined) opts.length = range.length
86 }
87
88 const stream = self._blobs.createReadStream(blobId, opts)
89 for await (const chunk of stream) {
90 yield chunk
91 }
92 })()
93 }
94
95 scheduleOrphan (hash) {
96 if (this._orphanQueue.has(hash)) return
97 const timeoutId = setTimeout(() => this._clearIfStillOrphan(hash), this._graceMs)
98 this._orphanQueue.set(hash, timeoutId)
99 }
100
101 cancelOrphan (hash) {
102 const timeoutId = this._orphanQueue.get(hash)
103 if (timeoutId) {
104 clearTimeout(timeoutId)
105 this._orphanQueue.delete(hash)
106 }
107 }
108
109 async _clearIfStillOrphan (hash) {
110 this._orphanQueue.delete(hash)
111 const stillReferenced = await this.isReferencedAnywhere(hash)
112 if (stillReferenced) return
113 await this.clear(hash)
114 }
115
116 async listAll () {
117 const results = []
118 for await (const entry of this._index.createReadStream({ gte: 'blob:', lt: 'blob:\xff' })) {
119 results.push({
120 hash: entry.key.replace('blob:', ''),
121 ...entry.value
122 })
123 }
124 return results
125 }
58126}
59127
60128export function parseRange (header, totalSize) {
test/unit/media-gc.test.js +114 -0
……@@ -0,0 +1,114 @@
1import test from 'brittle'
2import Corestore from 'corestore'
3import path from 'bare-path'
4import fs from 'bare-fs'
5import os from 'bare-os'
6import { MediaStore } from '../../lib/media-store.js'
7
8function tmpdir () {
9 const dir = path.join(os.tmpdir(), 'hn-gc-test-' + Date.now() + '-' + Math.random().toString(36).slice(2))
10 fs.mkdirSync(dir, { recursive: true })
11 return dir
12}
13
14function rmrf (dir) {
15 try { fs.rmSync(dir, { recursive: true, force: true }) } catch {}
16}
17
18test('GC - scheduleOrphan clears blob after grace period', async (t) => {
19 const dir = tmpdir()
20 try {
21 const store = new Corestore(dir)
22 const media = new MediaStore(store, {
23 graceMs: 100,
24 isReferencedAnywhere: async () => false
25 })
26 await media.ready()
27
28 const buf = Buffer.from('orphan data')
29 const hash = await media.put(buf, 'text/plain')
30 t.is(await media.has(hash), true)
31
32 media.scheduleOrphan(hash)
33
34 // Wait for grace period + buffer
35 await new Promise(r => setTimeout(r, 250))
36
37 t.is(await media.has(hash), false)
38 } finally {
39 rmrf(dir)
40 }
41})
42
43test('GC - cancelOrphan prevents deletion', async (t) => {
44 const dir = tmpdir()
45 try {
46 const store = new Corestore(dir)
47 const media = new MediaStore(store, {
48 graceMs: 200,
49 isReferencedAnywhere: async () => false
50 })
51 await media.ready()
52
53 const buf = Buffer.from('rescued data')
54 const hash = await media.put(buf, 'text/plain')
55
56 media.scheduleOrphan(hash)
57
58 // Cancel before grace period
59 await new Promise(r => setTimeout(r, 50))
60 media.cancelOrphan(hash)
61
62 // Wait past original grace period
63 await new Promise(r => setTimeout(r, 300))
64
65 t.is(await media.has(hash), true)
66 } finally {
67 rmrf(dir)
68 }
69})
70
71test('GC - scheduleOrphan skips if still referenced', async (t) => {
72 const dir = tmpdir()
73 try {
74 const store = new Corestore(dir)
75 const media = new MediaStore(store, {
76 graceMs: 100,
77 isReferencedAnywhere: async () => true
78 })
79 await media.ready()
80
81 const buf = Buffer.from('still in use')
82 const hash = await media.put(buf, 'text/plain')
83
84 media.scheduleOrphan(hash)
85 await new Promise(r => setTimeout(r, 250))
86
87 t.is(await media.has(hash), true)
88 } finally {
89 rmrf(dir)
90 }
91})
92
93test('GC - duplicate scheduleOrphan is idempotent', async (t) => {
94 const dir = tmpdir()
95 try {
96 const store = new Corestore(dir)
97 const media = new MediaStore(store, {
98 graceMs: 100,
99 isReferencedAnywhere: async () => false
100 })
101 await media.ready()
102
103 const buf = Buffer.from('double schedule')
104 const hash = await media.put(buf, 'text/plain')
105
106 media.scheduleOrphan(hash)
107 media.scheduleOrphan(hash)
108
109 await new Promise(r => setTimeout(r, 250))
110 t.is(await media.has(hash), false)
111 } finally {
112 rmrf(dir)
113 }
114})