OverviewFilesCommitsIssuesDocsReleasesPeers

P2P-Hiverelay commits / d1dc33f

feat(notify): watch exact outbox lanes
d1dc33fe13fec84c924ae8e937586d0be74169fb
by bigdestiny2 <33146784+bigdestiny2@users.noreply.github.com> · 2026-07-29T12:37:13.000Z

Changed files (11) +338 -61

docs/NAMESPACE.md +12 -3
……@@ -10,7 +10,11 @@ The OutboxLog is a relay-hosted, single-writer, append-only signed log — one "
1010
1111- A **namespace registry** (operator-owned) admits named namespaces, each with its own capability profile.
1212- Every record is labeled (`_ns`) and every outbox group binds to exactly one namespace — a group can never straddle two.
13- A **blind namespace** forbids plaintext fields: the relay stores only sealed ciphertext (blind sealing in `blind-seal.js`).
13- A **blind namespace** forbids plaintext application fields: data rows store
14 only sealed ciphertext (blind sealing in `blind-seal.js`). Three exact signed
15 control shapes are admitted without a sealed body: global `head`, atomic
16 `commit` authorization, and `lane-head {id,version,updatedAt}`. Unknown or
17 additional fields on those shapes are rejected.
1418- An **opaque-id takedown** primitive lets the operator mark individual records do-not-serve without learning anything about their content.
1519
1620It also fixed a fleet footgun: ENV-provisioned boxes had no way to register a namespace, so *every* append was refused `unknown namespace` — the `HIVERELAY_OUTBOXLOG_NAMESPACE` env var now seeds the registry (persisted config still wins).
……@@ -110,7 +114,9 @@ The operator's takedown is gated by the admin key and references records by opaq
110114
111115- **Writer authentication**: per-record ed25519 signatures, `_k` must equal `appId`, namespace inside the signature domain.
112116- **Operator allowlist**: unregistered namespaces reject at create, append, and verify time.
113- **Blind mode**: `BLIND_FORBIDDEN_FIELDS` hard-blocks plaintext fields on blind namespaces.
117- **Blind mode**: `BLIND_FORBIDDEN_FIELDS` hard-blocks plaintext fields on blind
118 namespaces. `head`, `commit`, and `lane-head` use exact field allowlists so
119 atomic control metadata works without becoming a plaintext smuggling channel.
114120- **Caps**: enforced per namespace — `maxOutboxes`, `maxEntriesPerOutbox`, `maxValueBytes`, `bytesPerDay` (each resolved as min of namespace cap and global fallback).
115121- **Takedown**: admin-key-only, opaque-id, do-not-serve.
116122
……@@ -145,7 +151,10 @@ flowchart LR
145151```
146152
147153- **Service Contract**: app releases never require relay updates — admission is config, not code.
148- **Notify Mode-2** composes with the co-resident outboxlog.
154- **Notify Mode-2** composes with the co-resident outboxlog. Global
155 `notify-feed-head` remains available; sender-owned virtual mailboxes use
156 `notify-outbox-lane`, which observes only `lane-head!<opaque tag>` and emits a
157 payload-free wake.
149158- **Privacy transports**: encrypted wake/head hints are exactly the bounded control messages the Tor/Nym privacy lanes carry; the log bodies themselves stay on native paths (or Tor bulk).
150159
151160## 7. Honest limits
docs/NOTIFY-OUTBOX-REVIEW.md +17 -0
……@@ -115,3 +115,20 @@ full-snapshot write would have dominated append latency at this state size.
115115- **Coalescing limiter is in-memory** (noted above): a relay restart inside a
116116 `minIntervalSeconds` window can emit one extra wake. Accepted for V1; persisting
117117 last-fire timestamps is trivial if it ever matters.
118
119## 2026-07-29 sender-owned lane addendum
120
121Pear Bots exposed a fanout problem in the original global-head composition: one
122sender outbox can multiplex many recipients, so watching `head!<writer>` wakes
123all of them. HiveRelay now also supports `notify-outbox-lane` with an opaque
12432-byte `source.lane`. Relay-node subscribes to the same OutboxLog event stream
125but forwards only the exact `lane-head!<lane>` mutation. Watch renewal replaces
126the prior subscription under the deterministic watch id and does not consume a
127second slot.
128
129Blind atomic commits required a corresponding namespace correction. Blind data
130rows still require an exact sealed body, while signed global head, commit
131authorization, and lane-head records are accepted only under exact control-field
132allowlists. Extra plaintext on a control row is rejected. The integration test
133uses the real signature verifier and durable atomic commit journal, asserts
134lane isolation/coalescing, and confirms the provider receives an empty wake.
docs/PUSH-NOTIFICATION-SERVICE-SPEC.md +12 -2
……@@ -148,6 +148,7 @@ Example capability profile:
148148 "providers": ["runtime", "apns", "fcm", "webpush"],
149149 "credential_modes": ["runtime-owned", "app-owned"],
150150 "modes": ["direct", "watch", "presence-fallback"],
151 "watch_sources": ["notify-feed-head", "notify-outbox-lane"],
151152 "payload": {
152153 "max_ciphertext_bytes": 3072,
153154 "plaintext_allowed": false,
……@@ -622,6 +623,14 @@ V1 watch source kinds:
622623|---|---|---|
623624| `hypercore-head` | Core key and length only | Head length increases. |
624625| `notify-feed-head` | Dedicated app notification feed key and length only | Head length increases. |
626| `notify-outbox-lane` | Sender OutboxLog key and opaque 32-byte lane tag | Exact signed `lane-head!<lane>` mutation advances. |
627
628`notify-outbox-lane` is the sender-owned mailbox form. The source is
629`{kind, key:<outbox signing key>, lane:<opaque tag>, start}`. The co-resident
630OutboxLog observer matches only that lane-head key, never the global writer
631head, so unrelated recipients are not fanned out. The resulting provider
632delivery has an empty `payloadCiphertext`; the application fetches and verifies
633the lane after wake. Deterministic watch ids may be renewed idempotently.
625634
626635Relay-side watch rules in v1 are intentionally primitive. They can detect new
627636append-only activity but cannot run app code or decrypt app messages. The app
……@@ -793,8 +802,9 @@ Output:
793802 "ok": true,
794803 "watchId": "32-byte-hex",
795804 "source": {
796 "kind": "hypercore-head",
797 "key": "32-byte-hex-core-key"
805 "kind": "notify-outbox-lane",
806 "key": "32-byte-hex-outbox-writer-key",
807 "lane": "32-byte-hex-opaque-lane"
798808 }
799809}
800810```
packages/core/core/capability-doc.js +2 -1
……@@ -547,6 +547,7 @@ function buildServicesProtocolProfile (relay, servicesEnabled) {
547547 providers: notifyLimits.providers || ['runtime', 'apns', 'fcm', 'webpush'],
548548 credential_modes: notifyLimits.credentialModes || ['runtime-owned', 'app-owned'],
549549 modes: notifyLimits.modes || ['direct', 'watch', 'presence-fallback'],
550 watch_sources: notifyLimits.watchSources || [],
550551 payload: {
551552 max_ciphertext_bytes: notifyLimits.maxCiphertextBytes || 3072,
552553 plaintext_allowed: false,
……@@ -557,7 +558,7 @@ function buildServicesProtocolProfile (relay, servicesEnabled) {
557558 // `max_devices_per_user_app` used to be advertised here but was never
558559 // enforced anywhere in the service — it is dropped rather than kept as
559560 // a promise the relay doesn't keep.
560 max_watches_per_receive_cap: notifyLimits.maxChannelsPerReceiveCap || 128,
561 max_watches_per_receive_cap: notifyLimits.maxWatchesPerReceiveCap || 128,
561562 default_channel_per_hour: notifyLimits.defaultChannelPerHour || 30
562563 },
563564 // Published so a client can tell a delivering relay from a running one.
packages/core/core/relay-node/index.js +9 -0
……@@ -1831,6 +1831,15 @@ export class RelayNode extends EventEmitter {
18311831 if (event && event.key === 'head!' + source.key && !event.replay) onChange(event)
18321832 })
18331833 })
1834 // Sender-owned outboxes multiplex recipients into opaque virtual
1835 // lanes. Watching the global outbox head would wake every recipient
1836 // for every message, so lane watches observe only the exact signed
1837 // `lane-head!<opaque-tag>` mutation committed for that recipient.
1838 notifyEntry.provider.attachWatchSource('notify-outbox-lane', (source, onChange) => {
1839 return outboxApp.subscribe(source.key, {}, (event) => {
1840 if (event && event.key === 'lane-head!' + source.lane && !event.replay) onChange(event)
1841 })
1842 })
18341843 }
18351844
18361845 // Set up seeded apps callback for catalog broadcast
packages/services/builtin/notify-service.js +15 -2
……@@ -45,6 +45,12 @@ export const NOTIFY_MODES = Object.freeze([
4545 'presence-fallback'
4646])
4747
48export const NOTIFY_WATCH_SOURCE_KINDS = Object.freeze([
49 'hypercore-head',
50 'notify-feed-head',
51 'notify-outbox-lane'
52])
53
4854const HEX_32 = /^[0-9a-f]{64}$/i
4955const HEX_16 = /^[0-9a-f]{32}$/i
5056const HEX_SIG = /^[0-9a-f]{128}$/i
……@@ -552,12 +558,13 @@ export class NotifyService extends ServiceProvider {
552558 verifySignedAny(params, NOTIFY_DOMAINS.watch, [params.signatureKey, receiveCap.user, receiveCap.device])
553559
554560 const source = objectOr(params.source, null)
555 if (!source || (source.kind !== 'hypercore-head' && source.kind !== 'notify-feed-head')) throw fail('SOURCE_DENIED', 'unsupported watch source')
561 if (!source || !NOTIFY_WATCH_SOURCE_KINDS.includes(source.kind)) throw fail('SOURCE_DENIED', 'unsupported watch source')
556562 // A watch is accepted only when an observer for the source kind is
557563 // actually attached — never bill for a watch that cannot fire. (#142)
558564 if (!this._watchSources.has(source.kind)) throw fail('SOURCE_UNAVAILABLE', 'no observer attached for watch source kind ' + source.kind)
559565 requireHex(source.key, 'source.key')
560 const existingForCap = [...this.watches.values()].filter(w => w.receiveCap === receiveCap.capId).length
566 if (source.kind === 'notify-outbox-lane') requireHex(source.lane, 'source.lane')
567 const existingForCap = [...this.watches.values()].filter(w => w.receiveCap === receiveCap.capId && w.watchId !== params.watchId).length
561568 if (existingForCap >= this.maxWatchesPerReceiveCap) throw fail('WATCH_LIMIT', 'receive capability watch limit reached')
562569
563570 const watch = freezeClone({
……@@ -572,6 +579,7 @@ export class NotifyService extends ServiceProvider {
572579 source: {
573580 kind: source.kind,
574581 key: source.key,
582 ...(source.kind === 'notify-outbox-lane' ? { lane: source.lane } : {}),
575583 start: Number.isSafeInteger(source.start) && source.start >= 0 ? source.start : 0
576584 },
577585 channel: params.channel,
……@@ -582,6 +590,10 @@ export class NotifyService extends ServiceProvider {
582590 signatureKey: params.signatureKey || receiveCap.user,
583591 lastSeq: Number.isSafeInteger(source.start) && source.start >= 0 ? source.start : 0
584592 })
593 // Deterministic watch ids let clients safely renew after a lost response.
594 // Replace the old observer before installing the new signed lease so a
595 // renewal never leaks a duplicate subscription or consumes another slot.
596 this._unsubscribeWatch(watch.watchId)
585597 this.watches.set(watch.watchId, watch)
586598 this._subscribeWatch(watch)
587599 await this._saveState()
……@@ -723,6 +735,7 @@ export class NotifyService extends ServiceProvider {
723735 providers: NOTIFY_PROVIDERS,
724736 credentialModes: NOTIFY_CREDENTIAL_MODES,
725737 modes: NOTIFY_MODES,
738 watchSources: [...this._watchSources.keys()].sort(),
726739 // The capability doc reads this to decide whether to advertise notify-v1.
727740 egress: pushEgressProfile(this.provider)
728741 }
packages/services/builtin/outboxlog/outbox-log.js +50 -6
……@@ -70,6 +70,20 @@ const BLIND_FORBIDDEN_FIELDS = new Set(['plaintext', 'plainText', 'cleartext', '
7070const BLIND_BODY_FIELDS = new Set(['sealed'])
7171const BLIND_SEAL_FIELDS = new Set(['version', 'alg', 'nonce', 'ciphertext', 'keyId'])
7272const BLIND_SEAL_ALG = /^[a-z0-9][a-z0-9._+-]{0,63}$/
73const BLIND_CONTROL_FIELDS = Object.freeze({
74 head: new Set(['id', 'version', 'count', 'root', 'updatedAt', ...SIG_FIELDS]),
75 commit: new Set([
76 'id',
77 'appId',
78 'expectedVersion',
79 'expectedRoot',
80 'mutationSigs',
81 'headSig',
82 'createdAt',
83 ...SIG_FIELDS
84 ]),
85 'lane-head': new Set(['id', 'version', 'updatedAt', ...SIG_FIELDS])
86})
7387
7488const hex = (n) => randomBytes(n).toString('hex')
7589
……@@ -214,7 +228,7 @@ export function createOutboxLog ({
214228 assertJournalMutationAllowed()
215229 if (!op || typeof op.type !== 'string' || op.type.length > 64 || !op.data || op.data.id == null) throw fail('bad op', 400)
216230 assertNoNestedSignatureFields(op.data)
217 const namespaceInfo = namespaceInfoForAppend(op.data)
231 const namespaceInfo = namespaceInfoForAppend(op.type, op.data)
218232 const id = String(op.data.id)
219233 if (id.length > maxIdLength) throw fail('id too long', 400)
220234
……@@ -1101,13 +1115,13 @@ export function createOutboxLog ({
11011115 return info
11021116 }
11031117
1104 function namespaceInfoForAppend (data) {
1118 function namespaceInfoForAppend (type, data) {
11051119 const ns = recordNamespace(data)
11061120 if (!ns && customVerifyAppend && !namespaceRegistry.configured) return null
11071121 if (!ns) throw fail('missing namespace', 400)
11081122 const info = namespaceRegistry.get(ns)
11091123 if (!info) throw fail('unknown namespace', 400)
1110 if (!namespaceRecordAllowed(info, data)) throw fail('namespace policy rejected record', 400)
1124 if (!namespaceRecordAllowed(info, type, data)) throw fail('namespace policy rejected record', 400)
11111125 return info
11121126 }
11131127
……@@ -2705,7 +2719,7 @@ export function verifyOutboxRecordSignature ({ appId, type, data }, { namespace
27052719 if (!isHex(driveKey, 64) || !isHex(sigHex, 128)) return false
27062720 if (algorithm !== 'ed25519') return false
27072721 if (!namespaceInfo) return false
2708 if (!namespaceRecordAllowed(namespaceInfo, data)) return false
2722 if (!namespaceRecordAllowed(namespaceInfo, type, data)) return false
27092723
27102724 const signed = `pear.app.${driveKey}:${ns}:${canonicalOutboxRecord(type, data)}`
27112725 const signature = b4a.from(sigHex, 'hex')
……@@ -2751,10 +2765,40 @@ function recordNamespace (data) {
27512765 return data && typeof data._ns === 'string' ? normalizeNamespaceName(data._ns) : null
27522766}
27532767
2754function namespaceRecordAllowed (namespaceInfo, data) {
2768function namespaceRecordAllowed (namespaceInfo, type, data) {
27552769 if (!namespaceInfo) return false
27562770 if (!namespaceInfo.blind) return true
2757 return isOutboxBlindRecord(data) && !hasBlindForbiddenField(data)
2771 if (hasBlindForbiddenField(data)) return false
2772 if (isOutboxBlindRecord(data)) return true
2773 return isBlindControlRecord(type, data)
2774}
2775
2776// Blind namespaces still need a small authenticated control plane. Atomic
2777// commits cannot work if their signed `head` and `commit` authorization rows
2778// are required to contain an application ciphertext body; targeted push also
2779// needs an opaque per-lane cursor. Admit only these three exact metadata shapes
2780// so the exception cannot become a plaintext side channel for application data.
2781function isBlindControlRecord (type, data) {
2782 const allowed = BLIND_CONTROL_FIELDS[type]
2783 if (!allowed || !data || typeof data !== 'object' || Array.isArray(data)) return false
2784 const keys = Object.keys(data)
2785 if (keys.some(key => !allowed.has(key))) return false
2786 if (type === 'head') {
2787 return typeof data.id === 'string' &&
2788 Number.isSafeInteger(data.version) && data.version >= 1 &&
2789 Number.isSafeInteger(data.count) && data.count >= 0 &&
2790 isHex(data.root, 64) && Number.isSafeInteger(data.updatedAt) && data.updatedAt >= 0
2791 }
2792 if (type === 'commit') {
2793 return typeof data.id === 'string' && typeof data.appId === 'string' &&
2794 Number.isSafeInteger(data.expectedVersion) && data.expectedVersion >= 0 &&
2795 isHex(data.expectedRoot, 64) && Array.isArray(data.mutationSigs) &&
2796 data.mutationSigs.every(sig => isHex(sig, 128)) &&
2797 isHex(data.headSig, 128) && Number.isSafeInteger(data.createdAt) && data.createdAt >= 0
2798 }
2799 return isHex(data.id, 64) &&
2800 Number.isSafeInteger(data.version) && data.version >= 1 &&
2801 Number.isSafeInteger(data.updatedAt) && data.updatedAt >= 0
27582802}
27592803
27602804export function createOutboxBlindSealedBody ({
test/unit/capability-doc.test.js +3 -1
……@@ -159,7 +159,7 @@ test('capability doc advertises notify and outboxlog service profiles when runni
159159 config: { enableServices: true },
160160 serviceRegistry: {
161161 services: new Map([
162 ['notify', { status: 'running', version: '0.1.0', provider: { limits: () => ({ providers: ['runtime', 'apns', 'fcm', 'webpush'], credentialModes: ['runtime-owned', 'app-owned'], modes: ['direct', 'watch', 'presence-fallback'], egress: { live: true } }) } }],
162 ['notify', { status: 'running', version: '0.1.0', provider: { limits: () => ({ providers: ['runtime', 'apns', 'fcm', 'webpush'], credentialModes: ['runtime-owned', 'app-owned'], modes: ['direct', 'watch', 'presence-fallback'], watchSources: ['notify-feed-head', 'notify-outbox-lane'], maxWatchesPerReceiveCap: 64, egress: { live: true } }) } }],
163163 ['outboxlog', { status: 'running', version: '0.1.0' }]
164164 ]),
165165 catalog: () => [
……@@ -174,6 +174,8 @@ test('capability doc advertises notify and outboxlog service profiles when runni
174174 t.ok(doc.features.includes('outboxlog-v1'))
175175 t.is(doc.protocol_profile.services.notify.version, '0.1.0')
176176 t.is(doc.protocol_profile.services.notify.payload.plaintext_allowed, false)
177 t.alike(doc.protocol_profile.services.notify.watch_sources, ['notify-feed-head', 'notify-outbox-lane'])
178 t.is(doc.protocol_profile.services.notify.limits.max_watches_per_receive_cap, 64)
177179 t.is(doc.protocol_profile.services.outboxlog.model, 'single-writer-signed-outbox')
178180})
179181
test/unit/notify-outboxlog-wake.test.js +130 -39
……@@ -1,6 +1,6 @@
11/**
2 * End-to-end wake loop: a mailbox record lands on an outbox → the head row
3 * bumps → the notify watch fires → a real push adapter signs an APNS request.
2 * End-to-end wake loops for both sender-owned virtual lanes and the legacy
3 * global-head compatibility watch.
44 *
55 * This exists because the loop was broken in a way no single-component test
66 * could see. The relay bridge only fires on a row keyed `head!<appId>`, and
……@@ -11,7 +11,7 @@
1111 */
1212
1313import test from 'brittle'
14import { generateKeyPairSync, createPublicKey, verify as cryptoVerify } from 'node:crypto'
14import { createHash, generateKeyPairSync, createPublicKey, verify as cryptoVerify } from 'node:crypto'
1515import b4a from 'b4a'
1616import sodium from 'sodium-universal'
1717import {
……@@ -19,7 +19,12 @@ import {
1919 NOTIFY_DOMAINS,
2020 notifySignaturePayload
2121} from '../../packages/services/builtin/notify-service.js'
22import { OutboxLogApp } from '../../packages/services/builtin/outboxlog/index.js'
22import {
23 OutboxLogApp,
24 canonicalOutboxRecord,
25 createMemoryOutboxJournal,
26 createOutboxBlindSealedBody
27} from '../../packages/services/builtin/outboxlog/index.js'
2328import { sealDeviceToken } from '../../packages/services/builtin/notify-push/index.js'
2429
2530const NOW = 1782864000000
……@@ -28,74 +33,102 @@ const PB_NAMESPACE = 'pear-bots'
2833
2934const tick = () => new Promise(resolve => setImmediate(resolve))
3035
31test('wake loop: a pear-bots mailbox enqueue wakes the recipient device', async (t) => {
36test('wake loop: a signed atomic pear-bots lane commit wakes only that opaque lane', async (t) => {
3237 const relay = keyPair(51)
3338 const user = keyPair(52)
3439 const device = keyPair(53)
3540 const sender = keyPair(54)
36 // pear-bots uses the recipient's own key as the outbox appId, and the same
37 // value as the notify watch source key. That equality is what closes the loop.
38 const ownerKey = hex(60)
41 const laneA = hex(60)
42 const laneB = hex(61)
3943
4044 const provider = { attempts: [], async send (d) { this.attempts.push(d); return { status: 'accepted_by_provider' } } }
41 const outbox = new OutboxLogApp({ verifyAppend: () => true, persistence: false })
42 // A relay only accepts records tagged `_ns: 'pear-bots'` if that namespace is
43 // registered. Without this, every enqueue is rejected 400 "unknown namespace"
44 // — so this line is a real operator prerequisite, not test scaffolding.
45 await outbox.start({ config: { outboxlog: { namespaces: { [PB_NAMESPACE]: { blind: false } } } } })
45 const outbox = new OutboxLogApp({
46 journal: createMemoryOutboxJournal([], { durableSync: true }),
47 persistence: false
48 })
49 await outbox.start({ config: { outboxlog: { namespaces: { [PB_NAMESPACE]: { blind: true } } } } })
4650 const notify = new NotifyService({ keyPair: relay, provider, clock: () => NOW })
4751
48 // Verbatim the closure relay-node/index.js installs at startup.
49 notify.attachWatchSource('notify-feed-head', (source, onChange) => {
52 // Verbatim the virtual-lane closure relay-node/index.js installs at startup.
53 notify.attachWatchSource('notify-outbox-lane', (source, onChange) => {
5054 return outbox.subscribe(source.key, {}, (event) => {
51 if (event && event.key === 'head!' + source.key && !event.replay) onChange(event)
55 if (event && event.key === 'lane-head!' + source.lane && !event.replay) onChange(event)
5256 })
5357 })
5458
55 await installHappyPath(notify, { relay, user, device, sender, modes: ['direct', 'watch'] })
56 const installed = await notify.watch(signed(user, NOTIFY_DOMAINS.watch, {
59 await installHappyPath(notify, { relay, user, device, sender, modes: ['watch'] })
60 const watchA = await notify.watch(signed(user, NOTIFY_DOMAINS.watch, {
5761 type: 'hiverelay.notify.watch.v1',
5862 watchId: hex(59),
5963 receiveCap: hex(6),
6064 sendCap: hex(7),
6165 app: hex(5),
6266 audience: relay.hex,
63 source: { kind: 'notify-feed-head', key: ownerKey, start: 0 },
67 source: { kind: 'notify-outbox-lane', key: sender.hex, lane: laneA, start: 0 },
6468 channel: 'message',
6569 policy: { minIntervalSeconds: 30 },
6670 createdAt: NOW,
6771 expiresAt: NOW + HOUR
6872 }))
69 t.is(installed.ok, true, 'watch installed against the recipient outbox')
70
71 outbox.create({ appId: ownerKey })
73 const watchB = await notify.watch(signed(user, NOTIFY_DOMAINS.watch, {
74 type: 'hiverelay.notify.watch.v1',
75 watchId: hex(58),
76 receiveCap: hex(6),
77 sendCap: hex(7),
78 app: hex(5),
79 audience: relay.hex,
80 source: { kind: 'notify-outbox-lane', key: sender.hex, lane: laneB, start: 0 },
81 channel: 'message',
82 policy: { minIntervalSeconds: 30 },
83 createdAt: NOW,
84 expiresAt: NOW + HOUR
85 }))
86 t.is(watchA.ok, true)
87 t.is(watchB.ok, true)
7288
73 // 1. The mailbox row alone — this is all pear-bots used to write.
74 outbox.append({
75 appId: ownerKey,
76 op: { type: 'mailbox', data: { id: 'c'.repeat(32), _ns: PB_NAMESPACE, payload: 'sealed-ciphertext', createdAt: NOW } }
89 const sealed = createOutboxBlindSealedBody({
90 nonce: b4a.toString(b4a.alloc(24, 1), 'base64url'),
91 ciphertext: b4a.toString(b4a.alloc(48, 2), 'base64url'),
92 keyId: 'epoch-1'
7793 })
78 await tick()
79 t.is(provider.attempts.length, 0, 'a mailbox row on its own wakes nobody — this was the whole bug')
80
81 // 2. The head row pear-bots now writes immediately after it.
82 outbox.append({
83 appId: ownerKey,
84 op: { type: 'head', data: { id: ownerKey, _ns: PB_NAMESPACE, bumpedAt: NOW } }
94 const first = atomicTransition(sender, {
95 expected: { version: 0, root: sha256('') },
96 mutations: [
97 { type: 'mailbox', fields: { id: laneA + '!op-1', body: sealed, expiresAt: NOW + HOUR } },
98 { type: 'lane-head', fields: { id: laneA, version: 1, updatedAt: NOW } }
99 ]
85100 })
101 outbox.commit(sender.hex, first.commit)
86102 await tick()
87 t.is(provider.attempts.length, 1, 'head row wakes the recipient device')
103 t.is(provider.attempts.length, 1, 'the exact lane A cursor wakes its watch')
88104
89105 const wake = provider.attempts[0]
90106 t.is(wake.payloadCiphertext, '', 'the wake is an opaque poke, not a message')
91 t.absent(JSON.stringify(wake).includes('sealed-ciphertext'), 'no record content reaches the push provider')
107 t.is(wake.watch.watchId, hex(59), 'lane B did not fan out')
108 t.absent(JSON.stringify(wake).includes(sealed.sealed.ciphertext), 'no record content reaches the push provider')
92109 t.absent('intent' in wake, 'watch wakes carry no intent — adapters must not require one')
93110
94 // 3. Coalescing: a second enqueue inside minIntervalSeconds must not
95 // double-spend the device's push budget.
96 outbox.append({ appId: ownerKey, op: { type: 'head', data: { id: ownerKey, _ns: PB_NAMESPACE, bumpedAt: NOW + 1000 } } })
111 const second = atomicTransition(sender, {
112 expected: { version: first.head.version, root: first.head.root },
113 base: first.census,
114 mutations: [
115 { type: 'mailbox', fields: { id: laneB + '!op-2', body: sealed, expiresAt: NOW + HOUR } },
116 { type: 'lane-head', fields: { id: laneB, version: 1, updatedAt: NOW } }
117 ]
118 })
119 outbox.commit(sender.hex, second.commit)
97120 await tick()
98 t.is(provider.attempts.length, 1, 'a second head bump inside the coalescing window is dropped')
121 t.is(provider.attempts.length, 2, 'lane B independently wakes its exact watch')
122 t.is(provider.attempts[1].watch.watchId, hex(58))
123
124 const third = atomicTransition(sender, {
125 expected: { version: second.head.version, root: second.head.root },
126 base: second.census,
127 mutations: [{ type: 'lane-head', fields: { id: laneA, version: 2, updatedAt: NOW + 1000 } }]
128 })
129 outbox.commit(sender.hex, third.commit)
130 await tick()
131 t.is(provider.attempts.length, 2, 'same-lane bumps coalesce inside minIntervalSeconds')
99132})
100133
101134test('wake loop: the wake reaches a real signing adapter', async (t) => {
……@@ -183,6 +216,64 @@ test('wake loop: the wake reaches a real signing adapter', async (t) => {
183216 await notify.stop()
184217})
185218
219function atomicTransition (writer, opts) {
220 const census = new Map(opts.base || [])
221 const mutations = opts.mutations.map(mutation => {
222 const data = signOutboxRecord(writer, mutation.type, mutation.fields)
223 const key = mutation.type + '!' + data.id
224 census.set(key, key + '\x00' + data._sig)
225 return { type: mutation.type, data }
226 })
227 const values = [...census.values()].sort()
228 const createdAt = opts.createdAt == null ? NOW : opts.createdAt
229 const head = signOutboxRecord(writer, 'head', {
230 id: writer.hex,
231 version: opts.expected.version + 1,
232 count: values.length,
233 root: sha256(values.join('\x01')),
234 updatedAt: createdAt
235 })
236 const fields = {
237 appId: writer.hex,
238 expectedVersion: opts.expected.version,
239 expectedRoot: opts.expected.root,
240 mutationSigs: mutations.map(mutation => mutation.data._sig),
241 headSig: head._sig,
242 createdAt
243 }
244 const commitId = sha256(canonicalOutboxRecord('commit-id', fields))
245 return {
246 head,
247 census,
248 commit: {
249 schema: 1,
250 commitId,
251 expected: { ...opts.expected },
252 mutations,
253 head: { type: 'head', data: head },
254 authorization: signOutboxRecord(writer, 'commit', { id: commitId, ...fields })
255 }
256 }
257}
258
259function signOutboxRecord (writer, type, fields) {
260 const data = {
261 ...fields,
262 _k: writer.hex,
263 _dk: writer.hex,
264 _ns: PB_NAMESPACE,
265 _alg: 'ed25519'
266 }
267 const message = 'pear.app.' + data._dk + ':' + data._ns + ':' + canonicalOutboxRecord(type, data)
268 const signature = b4a.alloc(64)
269 sodium.crypto_sign_detached(signature, b4a.from(message, 'utf8'), writer.secretKey)
270 return { ...data, _sig: b4a.toString(signature, 'hex') }
271}
272
273function sha256 (value) {
274 return createHash('sha256').update(String(value), 'utf8').digest('hex')
275}
276
186277async function installHappyPath (notify, { relay, user, device, sender, modes = ['direct'], tokenCiphertext = 'encrypted-provider-token' }) {
187278 const app = hex(5)
188279 const bindingId = hex(8)
test/unit/notify-service.test.js +39 -0
……@@ -63,6 +63,45 @@ test('notify service: signed direct wake path stores redacted delivery event', a
6363 t.ok(verifyNotifySignature(events.events[0], NOTIFY_DOMAINS.deliveryEvent, relay.hex))
6464})
6565
66test('notify service: lane watch preserves and validates the opaque lane selector', async (t) => {
67 const relay = keyPair(71)
68 const user = keyPair(72)
69 const device = keyPair(73)
70 const sender = keyPair(74)
71 const notify = new NotifyService({ keyPair: relay, clock: () => NOW })
72 notify.attachWatchSource('notify-outbox-lane', () => () => {})
73 await installHappyPath(notify, { relay, user, device, sender, modes: ['watch'] })
74
75 const lane = hex(75)
76 const installed = await notify.watch(signed(user, NOTIFY_DOMAINS.watch, {
77 type: 'hiverelay.notify.watch.v1',
78 watchId: hex(76),
79 receiveCap: hex(6),
80 sendCap: hex(7),
81 app: hex(5),
82 audience: relay.hex,
83 source: { kind: 'notify-outbox-lane', key: sender.hex, lane, start: 0 },
84 channel: 'message',
85 policy: { minIntervalSeconds: 30 },
86 createdAt: NOW,
87 expiresAt: NOW + HOUR
88 }))
89 t.alike(installed.source, { kind: 'notify-outbox-lane', key: sender.hex, lane, start: 0 })
90
91 await t.exception(notify.watch(signed(user, NOTIFY_DOMAINS.watch, {
92 type: 'hiverelay.notify.watch.v1',
93 watchId: hex(77),
94 receiveCap: hex(6),
95 sendCap: hex(7),
96 app: hex(5),
97 audience: relay.hex,
98 source: { kind: 'notify-outbox-lane', key: sender.hex, lane: 'short' },
99 channel: 'message',
100 createdAt: NOW,
101 expiresAt: NOW + HOUR
102 })), /source.lane must be 32-byte hex/)
103})
104
66105test('notify service: abuse limits are configurable and zero disables a scope (quota_exhausted)', async (t) => {
67106 const relay = keyPair(1)
68107 const user = keyPair(2)
test/unit/outboxlog-atomic-commit.test.js +49 -7
……@@ -71,6 +71,48 @@ test('outboxlog atomic commit: durable genesis is one journal transaction and re
7171 t.is(journal.entries().length, 1)
7272})
7373
74test('outboxlog atomic commit: blind namespace admits only sealed data and exact control rows', (t) => {
75 const writer = keyPair(68)
76 const journal = durableMemoryJournal()
77 const namespace = 'pear-bots'
78 const lane = 'a'.repeat(64)
79 const body = {
80 sealed: {
81 version: 1,
82 alg: 'xchacha20poly1305',
83 nonce: 'opaque-nonce',
84 ciphertext: 'opaque-ciphertext'
85 }
86 }
87 const log = createOutboxLog({
88 journal,
89 namespaces: { [namespace]: { blind: true } }
90 })
91 const built = transition(writer, {
92 namespace,
93 includeAuthor: false,
94 expected: { version: 0, root: EMPTY_ROOT },
95 mutations: [
96 { type: 'mailbox', fields: { id: lane + '!m1', body } },
97 { type: 'lane-head', fields: { id: lane, version: 1, updatedAt: 1000 } }
98 ]
99 })
100
101 const receipt = log.sync.commit(writer.publicKeyHex, built.commit)
102 t.is(receipt.durable, true)
103 t.is(log.sync.get(writer.publicKeyHex, 'mailbox!' + lane + '!m1').body.sealed.ciphertext, 'opaque-ciphertext')
104 t.is(log.sync.get(writer.publicKeyHex, 'lane-head!' + lane).version, 1)
105
106 const leaked = transition(writer, {
107 namespace,
108 includeAuthor: false,
109 expected: { version: 1, root: built.head.root },
110 base: built.census,
111 mutations: [{ type: 'lane-head', fields: { id: lane, version: 2, updatedAt: 2000, message: 'plaintext leak' } }]
112 })
113 t.is(throws(() => log.sync.commit(writer.publicKeyHex, leaked.commit)).status, 400, 'extra lane-head fields cannot bypass blind sealing')
114})
115
74116test('outboxlog signatures: nested reserved metadata is rejected even when signature verification is disabled', (t) => {
75117 const writer = keyPair(69)
76118 const journal = durableMemoryJournal()
……@@ -892,9 +934,9 @@ function transition (writer, opts) {
892934 const census = new Map(opts.base || [])
893935 const mutations = opts.mutations.map((mutation) => {
894936 const data = mutation.data || signRecord(writer, mutation.type, {
895 author: writer.publicKeyHex,
937 ...(opts.includeAuthor === false ? {} : { author: writer.publicKeyHex }),
896938 ...mutation.fields
897 })
939 }, opts.namespace)
898940 const key = mutation.type.replace(':', '!') + '!' + data.id
899941 census.set(key, key + '\x00' + data._sig)
900942 return { type: mutation.type, data, timestamp: '2026-07-10T00:00:00.000Z' }
……@@ -903,12 +945,12 @@ function transition (writer, opts) {
903945 const computedRoot = sha256(censusValues.join('\x01'))
904946 const head = signRecord(writer, 'head', {
905947 id: writer.publicKeyHex,
906 author: writer.publicKeyHex,
948 ...(opts.includeAuthor === false ? {} : { author: writer.publicKeyHex }),
907949 version: opts.headVersion === undefined ? expected.version + 1 : opts.headVersion,
908950 count: opts.headCount === undefined ? censusValues.length : opts.headCount,
909951 root: opts.headRoot || computedRoot,
910952 updatedAt: opts.createdAt || 1000
911 })
953 }, opts.namespace)
912954 const fields = {
913955 appId: writer.publicKeyHex,
914956 expectedVersion: expected.version,
……@@ -918,7 +960,7 @@ function transition (writer, opts) {
918960 createdAt: opts.createdAt || 1000
919961 }
920962 const commitId = sha256(canonicalOutboxRecord('commit-id', fields))
921 const authorization = signRecord(writer, 'commit', { id: commitId, ...fields })
963 const authorization = signRecord(writer, 'commit', { id: commitId, ...fields }, opts.namespace)
922964 return {
923965 head,
924966 census,
……@@ -972,12 +1014,12 @@ function keyPair (seedByte) {
9721014 return { publicKey, secretKey, publicKeyHex: b4a.toString(publicKey, 'hex') }
9731015}
9741016
975function signRecord (writer, type, fields) {
1017function signRecord (writer, type, fields, namespace = DEFAULT_OUTBOXLOG_NAMESPACE) {
9761018 const data = {
9771019 ...fields,
9781020 _k: writer.publicKeyHex,
9791021 _dk: 'd'.repeat(64),
980 _ns: DEFAULT_OUTBOXLOG_NAMESPACE,
1022 _ns: namespace,
9811023 _alg: 'ed25519'
9821024 }
9831025 const message = `pear.app.${data._dk}:${data._ns}:${canonicalOutboxRecord(type, data)}`