hypercompute / lib / node.js

// lib/node.js — ComputeNode: owns the corestore + durable-plane Autobase and
// the T0 (trusted, native) executor table. SPEC §2, §4.
//
// P0 scope: enrollment/transport are scaffolded (see enroll.js / transport.js)
// but the *core* — content-addressed submit -> execute -> result over the
// deterministic fold, with memoization — is real and exercised by test/e2e.js
// via replicateAndSync (no network).

const Corestore = require('corestore')
const b4a = require('b4a')
const crypto = require('hypercore-crypto')
const { openPool, taskId, resultKey, DEFAULT_LIMITS } = require('./pool')

// a leased claim is expired once the worker's own clock passes at + leaseMs.
// Reading the clock here (scheduling) is fine; apply() must never do this.
function claimExpired (claim) {
  if (!claim || typeof claim.at !== 'number' || typeof claim.leaseMs !== 'number') return true
  return Date.now() > claim.at + claim.leaseMs
}

// resolve `promise` unless `ms` elapses first (the wallMs bound). The underlying
// executor is not forcibly killed (a real WASM sandbox would be); we stop awaiting.
function withTimeout (promise, ms) {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('wallMs timeout')), ms)
    promise.then(
      (v) => { clearTimeout(timer); resolve(v) },
      (e) => { clearTimeout(timer); reject(e) }
    )
  })
}

class ComputeNode {
  constructor (store, bootstrap, { identity, executors } = {}) {
    this.store = store
    this.identity = identity || crypto.keyPair()
    this.executors = executors || {} // codeHash -> async (input) => output   (T0 native)
    this.base = openPool(store, bootstrap)
  }

  // storage: a Corestore instance, or a path string (corestore 7 / hypercore 11
  // is RocksDB-backed via hypercore-storage — there is no random-access-memory
  // backend in this era; tests/demos pass a temp dir).
  static async create (storage, bootstrap = null, opts = {}) {
    const store = storage instanceof Corestore ? storage : new Corestore(storage)
    const node = new ComputeNode(store, bootstrap, opts)
    await node.base.ready()
    return node
  }

  get key () { return this.base.key } // pool bootstrap key (share to join)
  get writerKey () { return this.base.local.key } // this node's autobase writer key
  get idHex () { return b4a.toString(this.identity.publicKey, 'hex') }

  // --- membership (P0: direct; real path is enroll.js blind-pairing) ----------
  async grantSelf (role = 'worker') {
    await this.base.append({ type: 'grant', identityPubKey: this.idHex, role })
  }

  async grant (identityPubKey, role = 'worker') {
    const hex = b4a.isBuffer(identityPubKey) ? b4a.toString(identityPubKey, 'hex') : identityPubKey
    await this.base.append({ type: 'grant', identityPubKey: hex, role })
  }

  async addWriter (writerKey, indexer = false) {
    const hex = b4a.isBuffer(writerKey) ? b4a.toString(writerKey, 'hex') : writerKey
    await this.base.append({ type: 'add-writer', writerKey: hex, indexer })
  }

  // --- content-addressed compute (the core, SPEC §4) --------------------------
  async lookup (codeHash, inputHash) {
    const node = await this.base.view.get(resultKey(codeHash, inputHash))
    return node ? node.value.output : null
  }

  // submit returns { output, memoHit:true } on a cache hit, else { id, memoHit:false }.
  // opts.limits ({ fuel, wallMs, memPages }) bound the task (SPEC §4.3); opts.tier
  // is the trust requirement (SPEC §4.4; v0.1 = T0).
  async submit (codeHash, inputHash, input, { limits, tier = 'T0' } = {}) {
    const cached = await this.lookup(codeHash, inputHash)
    if (cached !== null && cached !== undefined) return { output: cached, memoHit: true }
    const lim = limits || DEFAULT_LIMITS
    const id = taskId(codeHash, inputHash, lim)
    await this.base.append({ type: 'task', id, codeHash, inputHash, input, limits: lim, tier })
    return { id, memoHit: false }
  }

  // worker loop: execute every task that has a registered executor and no result
  // yet, respecting other workers' leased claims. Returns the number executed.
  async runPendingTasks () {
    let executed = 0
    const tasks = []
    for await (const { value: task } of this.base.view.createReadStream({ gte: 'task/', lt: 'task0' })) {
      tasks.push(task)
    }
    for (const task of tasks) {
      const existing = await this.base.view.get(resultKey(task.codeHash, task.inputHash))
      if (existing) continue
      const exec = this.executors[task.codeHash]
      if (!exec) continue

      // leased soft-lock (SPEC §6): skip a task another worker holds an UNEXPIRED
      // claim on. Best-effort dedup — simultaneous claims still race (wasted, not
      // corrupt: results are content-keyed + write-once). The clock is read HERE,
      // in the scheduler, never in apply() — the fold stays deterministic (SPEC §3).
      const claim = await this.base.view.get('claim/' + task.id)
      if (claim && claim.value.worker !== this.idHex && !claimExpired(claim.value)) continue

      const leaseMs = (task.limits && task.limits.wallMs) || DEFAULT_LIMITS.wallMs
      await this.base.append({ type: 'claim', id: task.id, worker: this.idHex, at: Date.now(), leaseMs })

      let output
      try {
        // wallMs is the one bound enforceable without a sandbox (anti-DoS, SPEC
        // §4.3); fuel/memPages are enforced by the WASM sandbox (P2).
        output = await withTimeout(exec(task.input), leaseMs) // T0 native execution
      } catch {
        continue // timeout or executor error — leave unexecuted for retry / another worker
      }
      await this.base.append({
        type: 'result',
        codeHash: task.codeHash,
        inputHash: task.inputHash,
        output,
        worker: this.idHex
      })
      executed++
    }
    return executed
  }

  close () { return this.base.close() }
}

module.exports = { ComputeNode }