Deepen supercomputer category with shared base and expanded APIs.

Adds supercomputer-base helpers, richer job/shard/steal/thermal modules,
priority jobs, claimMatching, and bumps all 14 modules to 0.3.1.

Co-authored-by: Cursor <[email protected]>

supercomputer/hyper-p2p-compute-shard/index.js

@@ -2,6 +2,7 @@ require('bare-process/global')
 const EventEmitter = require('bare-events')
 const b4a = require('b4a')
 const { initModuleSwarm, gossipSend } = require('../../_shared/p2p-bare.js')
+const { assertWorkId, clusterStats, shardProgress, normalizePeerId } = require('../../_shared/supercomputer-base.js')
 
 const PROTOCOL = 'compute-shard/v1'
 
@@ -12,7 +13,7 @@ const PROTOCOL = 'compute-shard/v1'
 class HyperP2PComputeShard extends EventEmitter {
   constructor (opts = {}) {
     super()
-    this._stats = { sharded: 0, completed: 0 }
+    this._stats = { sharded: 0, completed: 0, failed: 0, rebalanced: 0 }
     this.topic = opts.topic || null
     this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
     this.peerId = b4a.toString(this.keyPair.publicKey, 'hex')
@@ -23,9 +24,11 @@ class HyperP2PComputeShard extends EventEmitter {
   }
 
   shardWorkload (workId, chunks, peerIds = []) {
-    if (!workId) throw new Error('workId required')
+    const wid = assertWorkId(workId)
     if (!Array.isArray(chunks) || chunks.length < 1) throw new Error('chunks array required')
-    const peers = peerIds.length ? peerIds : (this.fabric ? this.fabric.listNodes().map((n) => n.peerId) : [this.peerId])
+    const peers = peerIds.length
+      ? peerIds.map((p) => normalizePeerId(p))
+      : (this.fabric ? this.fabric.listNodes().map((n) => n.peerId) : [this.peerId])
     if (!peers.length) throw new Error('no peers for sharding')
 
     const assignments = chunks.map((chunk, i) => ({
@@ -33,10 +36,11 @@ class HyperP2PComputeShard extends EventEmitter {
       peerId: peers[i % peers.length],
       chunk,
       state: 'pending',
-      result: null
+      result: null,
+      error: null
     }))
 
-    const plan = { workId: String(workId), assignments, createdAt: Date.now(), state: 'active' }
+    const plan = { workId: wid, assignments, createdAt: Date.now(), state: 'active' }
     this._plans.set(plan.workId, plan)
     this._stats.sharded++
     if (this._peerMsgs) gossipSend(this, { type: 'shard-plan', plan })
@@ -45,29 +49,85 @@ class HyperP2PComputeShard extends EventEmitter {
   }
 
   completeShard (workId, idx, result = null) {
-    const plan = this._plans.get(String(workId))
+    const plan = this._plans.get(assertWorkId(workId))
     if (!plan) return false
     const a = plan.assignments[idx]
     if (!a) return false
     a.state = 'done'
     a.result = result
+    this._maybeFinish(plan)
+    if (this._peerMsgs) gossipSend(this, { type: 'shard-done', workId: plan.workId, idx, result })
+    this.emit('shard-done', { workId: plan.workId, idx, result })
+    return true
+  }
+
+  failShard (workId, idx, error = 'shard failed') {
+    const plan = this._plans.get(assertWorkId(workId))
+    if (!plan) return false
+    const a = plan.assignments[idx]
+    if (!a) return false
+    a.state = 'failed'
+    a.error = String(error)
+    plan.state = 'failed'
+    this._stats.failed++
+    if (this._peerMsgs) gossipSend(this, { type: 'shard-fail', workId: plan.workId, idx, error: a.error })
+    this.emit('shard-fail', { workId: plan.workId, idx, error: a.error })
+    return true
+  }
+
+  _maybeFinish (plan) {
     const allDone = plan.assignments.every((x) => x.state === 'done')
     if (allDone) {
       plan.state = 'done'
       this._stats.completed++
       this.emit('work-complete', plan)
     }
-    if (this._peerMsgs) gossipSend(this, { type: 'shard-done', workId, idx, result })
-    this.emit('shard-done', { workId, idx, result })
-    return true
+  }
+
+  progress (workId) {
+    const plan = this._plans.get(assertWorkId(workId))
+    if (!plan) return 0
+    return shardProgress(plan.assignments)
+  }
+
+  pendingForPeer (peerId) {
+    const id = normalizePeerId(peerId)
+    const out = []
+    for (const plan of this._plans.values()) {
+      for (const a of plan.assignments) {
+        if (a.peerId === id && a.state === 'pending') out.push({ workId: plan.workId, ...a })
+      }
+    }
+    return out
+  }
+
+  rebalance (workId, peerIds = []) {
+    const plan = this._plans.get(assertWorkId(workId))
+    if (!plan || plan.state === 'done') return null
+    const peers = peerIds.length
+      ? peerIds.map((p) => normalizePeerId(p))
+      : (this.fabric ? this.fabric.listNodes().map((n) => n.peerId) : [this.peerId])
+    let i = 0
+    for (const a of plan.assignments) {
+      if (a.state === 'pending') {
+        a.peerId = peers[i % peers.length]
+        i++
+      }
+    }
+    this._stats.rebalanced++
+    this.emit('rebalance', plan)
+    return plan
   }
 
   getPlan (workId) {
-    const p = this._plans.get(String(workId))
+    const p = this._plans.get(assertWorkId(workId))
     return p ? { ...p, assignments: p.assignments.map((a) => ({ ...a })) } : null
   }
 
-  listPlans () { return [...this._plans.values()] }
+  listPlans (state = null) {
+    const all = [...this._plans.values()]
+    return state ? all.filter((p) => p.state === state) : all
+  }
 
   async ready () {
     if (this.swarm || !this.topic) return this
@@ -76,13 +136,17 @@ class HyperP2PComputeShard extends EventEmitter {
       onmessage: (data) => {
         if (data?.type === 'shard-plan' && data.plan) this._plans.set(data.plan.workId, data.plan)
         else if (data?.type === 'shard-done') this.completeShard(data.workId, data.idx, data.result)
+        else if (data?.type === 'shard-fail') this.failShard(data.workId, data.idx, data.error)
       }
     })
     return this
   }
 
   getStats () {
-    return { ...this._stats, plans: this._plans.size, protocol: PROTOCOL }
+    return clusterStats(this._stats, PROTOCOL, {
+      plans: this._plans.size,
+      active: this.listPlans('active').length
+    })
   }
 
   async close () {

supercomputer/hyper-p2p-compute-shard/package.json

@@ -1,6 +1,6 @@
 {
   "name": "hyper-p2p-compute-shard",
-  "version": "0.3.0",
+  "version": "0.3.1",
   "description": "Workload sharding for cluster supercomputer registry for Bare/Pear P2P supercomputer mesh.",
   "main": "index.js",
   "type": "commonjs",

supercomputer/hyper-p2p-compute-shard/test/test.js

@@ -15,3 +15,17 @@ test('compute-shard: shard across peers', async (t) => {
   await shard.close()
   await fabric.close()
 })
+
+test('progress fail rebalance', async (t) => {
+  const shard = new HyperP2PComputeShard()
+  shard.shardWorkload('w2', ['a', 'b'], ['p1', 'p2'])
+  t.is(shard.progress('w2'), 0)
+  shard.completeShard('w2', 0, 1)
+  t.is(shard.progress('w2'), 0.5)
+  shard.failShard('w2', 1, 'oom')
+  t.is(shard.getPlan('w2').state, 'failed')
+  shard.shardWorkload('w3', [1, 2], ['x'])
+  shard.rebalance('w3', ['y', 'z'])
+  t.ok(shard.pendingForPeer('y').length >= 0)
+  await shard.close()
+})