OverviewFilesCommitsIssuesDocsReleasesPeers

hypercompute commits / d720136

fix: leased claim soft-lock + task limits/tier (SPEC 4.1/4.3/6)
Two coupled P0 audit fixes (the lease duration IS the task's wallMs limit, so they share the task-lifecycle code and a test). claim was dead state — written by runPendingTasks but never read. Now it is a LEASED soft-lock that is consulted: a worker skips a task another worker holds an unexpired claim on (best-effort dedup; simultaneous claims still race, which is wasted not corrupt since results are content-keyed + write-once). The lease (at, leaseMs) is stamped by the worker; apply() only stores it and never reads the clock, so the fold stays deterministic (SPEC 3). A crashed claimant's lease expires -> the task recovers. tasks had no limits/tier, diverging from SPEC 4.1 (id = hash(code | input | limits)). Now submit takes { limits, tier }; tasks carry both; taskId folds limits in (a different budget is a different task) while the result stays keyed by (codeHash, inputHash) only. wallMs is enforced now as a real timeout (anti-DoS, SPEC 4.3); fuel/memPages are stored for the WASM sandbox (P2). test/claim-and-limits.js: limits-in-task-id + recorded on the op; unexpired foreign claim blocks; expired claim recovers. 6/6 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
d720136a336fb18274952e13d953e2f630811b16
by plankatron <fleeky@gmail.com> · 2026-06-18T00:02:43.000Z

Changed files (3) +129 -17

lib/node.js +48 -9
……@@ -9,7 +9,26 @@
99const Corestore = require('corestore')
1010const b4a = require('b4a')
1111const crypto = require('hypercore-crypto')
12const { openPool, taskId, resultKey } = require('./pool')
12const { openPool, taskId, resultKey, DEFAULT_LIMITS } = require('./pool')
13
14// a leased claim is expired once the worker's own clock passes at + leaseMs.
15// Reading the clock here (scheduling) is fine; apply() must never do this.
16function claimExpired (claim) {
17 if (!claim || typeof claim.at !== 'number' || typeof claim.leaseMs !== 'number') return true
18 return Date.now() > claim.at + claim.leaseMs
19}
20
21// resolve `promise` unless `ms` elapses first (the wallMs bound). The underlying
22// executor is not forcibly killed (a real WASM sandbox would be); we stop awaiting.
23function withTimeout (promise, ms) {
24 return new Promise((resolve, reject) => {
25 const timer = setTimeout(() => reject(new Error('wallMs timeout')), ms)
26 promise.then(
27 (v) => { clearTimeout(timer); resolve(v) },
28 (e) => { clearTimeout(timer); reject(e) }
29 )
30 })
31}
1332
1433class ComputeNode {
1534 constructor (store, bootstrap, { identity, executors } = {}) {
……@@ -54,17 +73,20 @@ class ComputeNode {
5473 return node ? node.value.output : null
5574 }
5675
57 // submit returns { output, memoHit:true } on a cache hit, else { id, memoHit:false }
58 async submit (codeHash, inputHash, input) {
76 // submit returns { output, memoHit:true } on a cache hit, else { id, memoHit:false }.
77 // opts.limits ({ fuel, wallMs, memPages }) bound the task (SPEC §4.3); opts.tier
78 // is the trust requirement (SPEC §4.4; v0.1 = T0).
79 async submit (codeHash, inputHash, input, { limits, tier = 'T0' } = {}) {
5980 const cached = await this.lookup(codeHash, inputHash)
6081 if (cached !== null && cached !== undefined) return { output: cached, memoHit: true }
61 const id = taskId(codeHash, inputHash)
62 await this.base.append({ type: 'task', id, codeHash, inputHash, input })
82 const lim = limits || DEFAULT_LIMITS
83 const id = taskId(codeHash, inputHash, lim)
84 await this.base.append({ type: 'task', id, codeHash, inputHash, input, limits: lim, tier })
6385 return { id, memoHit: false }
6486 }
6587
66 // worker loop: execute every task that has a registered executor and no result yet.
67 // Returns the number of tasks executed.
88 // worker loop: execute every task that has a registered executor and no result
89 // yet, respecting other workers' leased claims. Returns the number executed.
6890 async runPendingTasks () {
6991 let executed = 0
7092 const tasks = []
……@@ -76,8 +98,25 @@ class ComputeNode {
7698 if (existing) continue
7799 const exec = this.executors[task.codeHash]
78100 if (!exec) continue
79 await this.base.append({ type: 'claim', id: task.id, worker: this.idHex })
80 const output = await exec(task.input) // T0 native execution
101
102 // leased soft-lock (SPEC §6): skip a task another worker holds an UNEXPIRED
103 // claim on. Best-effort dedup — simultaneous claims still race (wasted, not
104 // corrupt: results are content-keyed + write-once). The clock is read HERE,
105 // in the scheduler, never in apply() — the fold stays deterministic (SPEC §3).
106 const claim = await this.base.view.get('claim/' + task.id)
107 if (claim && claim.value.worker !== this.idHex && !claimExpired(claim.value)) continue
108
109 const leaseMs = (task.limits && task.limits.wallMs) || DEFAULT_LIMITS.wallMs
110 await this.base.append({ type: 'claim', id: task.id, worker: this.idHex, at: Date.now(), leaseMs })
111
112 let output
113 try {
114 // wallMs is the one bound enforceable without a sandbox (anti-DoS, SPEC
115 // §4.3); fuel/memPages are enforced by the WASM sandbox (P2).
116 output = await withTimeout(exec(task.input), leaseMs) // T0 native execution
117 } catch {
118 continue // timeout or executor error — leave unexecuted for retry / another worker
119 }
81120 await this.base.append({
82121 type: 'result',
83122 codeHash: task.codeHash,
lib/pool.js +25 -8
……@@ -3,8 +3,8 @@
33// One optimistic Autobase whose view is a Hyperbee. Mutable state = a
44// deterministic fold (`apply`) over the append-only op log. View keys:
55// member/<idPub> -> { role }
6// task/<id> -> task op
7// claim/<id> -> { worker } (optimistic soft-lock)
6// task/<id> -> task op { codeHash, inputHash, input, limits, tier }
7// claim/<id> -> { worker, at, leaseMs } (leased soft-lock)
88// result/<codeHash>/<input> -> result op (the memoized chunk)
99//
1010// apply() MUST stay deterministic (no Date.now/random/external reads) — SPEC §3.
……@@ -18,9 +18,24 @@ function hash (str) {
1818 return b4a.toString(crypto.hash(b4a.from(str)), 'hex')
1919}
2020
21// content-addressed identity of a unit of work (SPEC §4.1)
22function taskId (codeHash, inputHash) {
23 return hash(codeHash + ':' + inputHash)
21// resource bounds (SPEC §4.3). fuel/memPages are enforced by the WASM sandbox
22// (P2); wallMs is enforceable today as a wall-clock timeout (see node.js).
23const DEFAULT_LIMITS = { fuel: 1000000000, wallMs: 30000, memPages: 16384 }
24
25// byte-stable encoding of limits (fixed field order) so the task id is identical
26// on every peer (content-addressing requires it).
27function encodeLimits (limits) {
28 const l = limits || DEFAULT_LIMITS
29 return l.fuel + '|' + l.wallMs + '|' + l.memPages
30}
31
32// content-addressed identity of a unit of work (SPEC §4.1:
33// id = hash(codeHash ‖ inputHash ‖ limits)). limits ARE part of the identity — a
34// different budget is a different task — but the RESULT is keyed by
35// (codeHash, inputHash) only: the same code+input yields the same output
36// regardless of budget.
37function taskId (codeHash, inputHash, limits) {
38 return hash(codeHash + ':' + inputHash + ':' + encodeLimits(limits))
2439}
2540
2641function resultKey (codeHash, inputHash) {
……@@ -53,8 +68,10 @@ function openPool (store, bootstrap) {
5368 case 'task': // content-addressed work unit
5469 await view.put('task/' + op.id, op)
5570 break
56 case 'claim': // optimistic soft-lock to dedupe workers
57 await view.put('claim/' + op.id, { worker: op.worker })
71 case 'claim': // leased soft-lock to dedupe workers (SPEC §6).
72 // apply only STORES the lease fields the worker stamped — it never
73 // reads the clock itself, so the fold stays deterministic (SPEC §3).
74 await view.put('claim/' + op.id, { worker: op.worker, at: op.at, leaseMs: op.leaseMs })
5875 break
5976 case 'result': { // memoized output, keyed by content — WRITE-ONCE
6077 // The fold is deterministic across peers, so "first result in
……@@ -72,4 +89,4 @@ function openPool (store, bootstrap) {
7289 })
7390}
7491
75module.exports = { openPool, taskId, resultKey, hash }
92module.exports = { openPool, taskId, resultKey, hash, DEFAULT_LIMITS }
test/claim-and-limits.js +56 -0
……@@ -0,0 +1,56 @@
1// test/claim-and-limits.js — two P0 audit fixes:
2// (1) tasks carry limits + tier, and limits are part of the task identity
3// (SPEC §4.1) while the result stays keyed by (codeHash, inputHash).
4// (2) the `claim` op is a LEASED soft-lock that is actually CONSULTED before
5// executing (it was previously written but never read — dead state).
6
7const test = require('brittle')
8const Corestore = require('corestore')
9const tmp = require('test-tmp')
10const { ComputeNode } = require('../lib/node')
11const { taskId } = require('../lib/pool')
12
13const codeHash = 'square-v1'
14const inputHash = 'abc123'
15const FOREIGN = 'ff'.repeat(32) // a worker id that is not ours
16const square = { [codeHash]: async (i) => i.n * i.n }
17
18test('limits are part of the task identity (SPEC §4.1), recorded on the task op', async (t) => {
19 const a = taskId(codeHash, inputHash, { fuel: 1, wallMs: 1, memPages: 1 })
20 const b = taskId(codeHash, inputHash, { fuel: 2, wallMs: 1, memPages: 1 })
21 t.not(a, b, 'a different budget is a different task id')
22
23 const node = await ComputeNode.create(await tmp(t), null, { executors: square })
24 await node.grantSelf('owner')
25 const sub = await node.submit(codeHash, inputHash, { n: 5 }, { limits: { fuel: 7, wallMs: 7, memPages: 7 }, tier: 'T1' })
26 const taskNode = await node.base.view.get('task/' + sub.id)
27 t.alike(taskNode.value.limits, { fuel: 7, wallMs: 7, memPages: 7 }, 'task records its limits')
28 t.is(taskNode.value.tier, 'T1', 'task records its tier')
29 await node.close()
30})
31
32test('an UNEXPIRED foreign claim blocks execution (claim is consulted, not dead state)', async (t) => {
33 const node = await ComputeNode.create(await tmp(t), null, { executors: square })
34 await node.grantSelf('owner')
35 const sub = await node.submit(codeHash, inputHash, { n: 6 })
36 // another worker holds a fresh lease on this task
37 await node.base.append({ type: 'claim', id: sub.id, worker: FOREIGN, at: Date.now(), leaseMs: 60000 })
38
39 const n = await node.runPendingTasks()
40 t.is(n, 0, 'task skipped while another worker holds an unexpired claim')
41 t.is(await node.lookup(codeHash, inputHash), null, 'no result produced')
42 await node.close()
43})
44
45test('an EXPIRED foreign claim does NOT block execution (lease recovery)', async (t) => {
46 const node = await ComputeNode.create(await tmp(t), null, { executors: square })
47 await node.grantSelf('owner')
48 const sub = await node.submit(codeHash, inputHash, { n: 6 })
49 // a stale lease: claimed far in the past, lease already elapsed
50 await node.base.append({ type: 'claim', id: sub.id, worker: FOREIGN, at: Date.now() - 120000, leaseMs: 60000 })
51
52 const n = await node.runPendingTasks()
53 t.is(n, 1, 'expired claim is ignored; task re-executed')
54 t.is(await node.lookup(codeHash, inputHash), 36, 'result produced (6*6)')
55 await node.close()
56})