| … | … | @@ -145,6 +145,89 @@ export async function resolveRelease (opts = {}) { |
| 145 | 145 | * sidecar, `pgh app install`, and its own next resolve. Polling is hourly by |
| 146 | 146 | * default, so the reconnect cost is irrelevant beside that. |
| 147 | 147 | */ |
| 148 | /** |
| 149 | * WATCH a registry for releases of one app, instead of asking it repeatedly. |
| 150 | * |
| 151 | * ⛔ WHY THIS EXISTS. `resolveRelease` opens a Corestore and a Hyperswarm, reads, and tears both |
| 152 | * down — so nothing survives to emit anything, and every caller is forced to poll. Worse, a one-shot |
| 153 | * read has to GUESS how long replication takes: the code above waits `settleMs` because, in its own |
| 154 | * words, "flush returns when peers are connected, not when the log has caught up". That guess was |
| 155 | * 500ms, which is correct for a warm replica and far too short for a cold one — measured on a |
| 156 | * handset as `resolveRelease` returning null and OTA reporting "no published release with a |
| 157 | * serverKey", which reads like a publishing failure and is not. |
| 158 | * |
| 159 | * ★ A held replica has no settle to guess. The view is built by autobase's apply as blocks arrive, |
| 160 | * and autobase emits `update` when it does — so you are TOLD when the data lands, however long it |
| 161 | * takes. That removes the entire class of "you did not wait long enough", rather than retuning it. |
| 162 | * |
| 163 | * Event-driven plus reconciliation, the same shape hyper-social #68 §2 landed for delivery: the |
| 164 | * event sets the latency floor, and the initial read after flush is the reconciliation that covers |
| 165 | * a replica which was already caught up before anyone subscribed. |
| 166 | * |
| 167 | * @returns {{ close: function, latest: function }} `latest()` is the last entry seen, or null. |
| 168 | */ |
| 169 | export async function watchLatestRelease (opts = {}) { |
| 170 | const { registryKeyHex, name, onRelease = () => {} } = opts |
| 171 | if (!registryKeyHex || registryKeyHex.length !== 64) { |
| 172 | throw new Error('watchLatestRelease requires a 64-char registryKeyHex') |
| 173 | } |
| 174 | if (!name) throw new Error('watchLatestRelease requires a name') |
| 175 | |
| 176 | const Corestore = (await import('corestore')).default |
| 177 | const Hyperswarm = (await import('hyperswarm')).default |
| 178 | |
| 179 | const dir = opts.storeDir || registryStoreDir(registryKeyHex) |
| 180 | fs.mkdirSync(dir, { recursive: true }) |
| 181 | |
| 182 | const store = new Corestore(dir) |
| 183 | const registry = new ReleaseRegistry(store, { bootstrap: b4a.from(registryKeyHex, 'hex') }) |
| 184 | await registry.ready() |
| 185 | |
| 186 | const swarm = new Hyperswarm() |
| 187 | swarm.on('connection', (conn) => store.replicate(conn)) |
| 188 | // client-only: we are reading someone else's registry, not serving ours. |
| 189 | swarm.join(registry._base.discoveryKey, { server: false, client: true }) |
| 190 | |
| 191 | let last = null |
| 192 | let closed = false |
| 193 | |
| 194 | // ⚠ Reports only CHANGES. A view can update many times while one release replicates, and a caller |
| 195 | // that rebuilt a fetcher on every tick would thrash for no reason. |
| 196 | const reread = async () => { |
| 197 | if (closed) return |
| 198 | try { |
| 199 | await registry.update() |
| 200 | const entry = await registry.latest(name) |
| 201 | if (!entry) return |
| 202 | if (last && last.treeHash === entry.treeHash) return |
| 203 | last = entry |
| 204 | await onRelease(entry) |
| 205 | } catch { /* a registry that cannot be read right now is not an error; the next update retries */ } |
| 206 | } |
| 207 | |
| 208 | registry._base.on('update', () => { reread().catch(() => {}) }) |
| 209 | |
| 210 | // Reconciliation: a replica already caught up before we subscribed emits nothing, so read once. |
| 211 | // NOT awaited on the swarm — `flush()` can take as long as a DHT lookup, and a caller must not |
| 212 | // block boot on it. The `update` event covers whatever arrives afterwards. |
| 213 | swarm.flush().then(reread).catch(() => {}) |
| 214 | await reread() |
| 215 | |
| 216 | return { |
| 217 | latest: () => last, |
| 218 | close: async () => { |
| 219 | closed = true |
| 220 | try { registry._base.removeAllListeners('update') } catch {} |
| 221 | // ⛔ CLOSE WHAT WE OPENED, all of it. A swarm left running is a DHT node and a set of sockets |
| 222 | // held awake by something the user switched off — the same leak shape that reached ~205k |
| 223 | // sockets on stortron because the thing that created a handle was not the thing that owned it. |
| 224 | try { await swarm.destroy() } catch {} |
| 225 | try { await registry.close() } catch {} |
| 226 | try { await store.close() } catch {} |
| 227 | } |
| 228 | } |
| 229 | } |
| 230 | |
| 148 | 231 | export function makeGetLatestRelease ({ registryKeyHex, preferSidecar, settleMs, storeDir } = {}) { |
| 149 | 232 | return async function getLatestRelease (name) { |
| 150 | 233 | const { entry, cleanup } = await resolveRelease({ |