aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/app.js5
-rw-r--r--src/ark/manager.js82
-rw-r--r--src/ark/proto.js86
3 files changed, 134 insertions, 39 deletions
diff --git a/src/app.js b/src/app.js
index 61b453b..efb0bc4 100644
--- a/src/app.js
+++ b/src/app.js
@@ -42,6 +42,7 @@ function stopArk() {
arkInitGen++;
if (arkTimer) clearInterval(arkTimer);
arkTimer = null;
+ if (ark) ark.stopMailboxStream();
ark = null;
arkConnectPromise = null;
}
@@ -89,7 +90,9 @@ function connectArk() {
ark = mgr;
const tick = () => mgr.sync().catch(() => {});
tick();
- // Regtest rounds are seconds apart; on mainnet a slower poll is plenty.
+ // Receives arrive in real time over the mailbox stream; the poll is the
+ // fallback and what drives in-flight boards/refreshes forward.
+ mgr.startMailboxStream();
arkTimer = setInterval(() => { if (ark === mgr) tick(); }, getNetwork() === 'regtest' ? 5000 : 30000);
render();
return mgr;
diff --git a/src/ark/manager.js b/src/ark/manager.js
index c2fd0c6..bb7eb5f 100644
--- a/src/ark/manager.js
+++ b/src/ark/manager.js
@@ -20,6 +20,7 @@ import * as musig2 from '@scure/btc-signer/musig2';
import {
getArkInfo, handshake, encodeAddress, decodeAddress, blindMailboxId,
readMailbox, decodeVtxo, arkIdFromServerPubkey, GrpcError,
+ grpcStream, decodeMailboxMessage, mailboxRequestBytes,
} from './proto.js';
import {
buildArkoorSend, cosignWithServer, buildSignedVtxoBytes,
@@ -145,31 +146,72 @@ export class ArkManager {
async sync() {
const mailbox = this._mailboxKey();
const { messages } = await readMailbox(this.arkUrl, mailbox, this.state.mailboxCheckpoint);
- const ourKeys = [hex.encode(this._key(0).pubkey)];
let changed = false;
- for (const m of messages) {
- if (m.checkpoint > this.state.mailboxCheckpoint) this.state.mailboxCheckpoint = m.checkpoint;
- if (m.kind !== 'arkoor') { changed = true; continue; }
- for (const v of m.vtxos) {
- if (this._vtxo(v.id)) continue;
- try {
- await validateVtxo(v, { serverPubkey: this.serverPub, chain: this.chain, expectPubkeys: ourKeys });
- } catch (e) {
- if (e instanceof VtxoValidationError) {
- this._movement({ type: 'receive', amountSat: v.amountSat, status: 'rejected', detail: e.message });
- changed = true;
- continue;
- }
- throw e; // network errors etc: retry next sync
+ for (const m of messages) changed = (await this._processMailboxMessage(m)) || changed;
+ if (changed) this._save();
+ await this.resumePending();
+ }
+
+ // Handle one mailbox message (from a poll or the live stream). Returns
+ // whether state changed; the caller saves. Duplicate-safe: vtxos dedupe by
+ // id and the checkpoint only moves forward.
+ async _processMailboxMessage(m) {
+ let changed = false;
+ if (m.checkpoint > this.state.mailboxCheckpoint) { this.state.mailboxCheckpoint = m.checkpoint; changed = true; }
+ if (m.kind !== 'arkoor') return changed;
+ const ourKeys = [hex.encode(this._key(0).pubkey)];
+ for (const v of m.vtxos) {
+ if (this._vtxo(v.id)) continue;
+ try {
+ await validateVtxo(v, { serverPubkey: this.serverPub, chain: this.chain, expectPubkeys: ourKeys });
+ } catch (e) {
+ if (e instanceof VtxoValidationError) {
+ this._movement({ type: 'receive', amountSat: v.amountSat, status: 'rejected', detail: e.message });
+ changed = true;
+ continue;
}
- this._addVtxo(v, v._raw.bytes, 0);
- this._movement({ type: 'receive', amountSat: v.amountSat, status: 'complete', vtxoId: v.id });
- changed = true;
+ throw e; // network errors etc: retry next sync
}
+ this._addVtxo(v, v._raw.bytes, 0);
+ this._movement({ type: 'receive', amountSat: v.amountSat, status: 'complete', vtxoId: v.id });
changed = true;
}
- if (changed) this._save();
- await this.resumePending();
+ return changed;
+ }
+
+ // Live mailbox push via the SubscribeMailbox gRPC stream — receives land the
+ // moment the sender delivers, instead of on the next poll. Reconnects with a
+ // short backoff until stopMailboxStream().
+ startMailboxStream() {
+ if (this._streaming) return;
+ this._streaming = true;
+ const loop = async () => {
+ while (this._streaming) {
+ try {
+ this._streamAbort = new AbortController();
+ await grpcStream(
+ this.arkUrl,
+ 'mailbox_server.MailboxService/SubscribeMailbox',
+ mailboxRequestBytes(this._mailboxKey(), this.state.mailboxCheckpoint),
+ {
+ signal: this._streamAbort.signal,
+ onMessage: (bytes) => {
+ this._processMailboxMessage(decodeMailboxMessage(bytes))
+ .then((changed) => { if (changed) this._save(); })
+ .catch(() => {}); // validation retried by the next poll
+ },
+ },
+ );
+ } catch {}
+ if (this._streaming) await new Promise((r) => setTimeout(r, 3000));
+ }
+ };
+ loop();
+ }
+ stopMailboxStream() {
+ this._streaming = false;
+ try { this._streamAbort?.abort(); } catch {}
+ this._streamAbort = null;
}
async resumePending() {
diff --git a/src/ark/proto.js b/src/ark/proto.js
index 24fb5a6..728686b 100644
--- a/src/ark/proto.js
+++ b/src/ark/proto.js
@@ -116,6 +116,47 @@ export async function grpcCall(base, method, reqBytes) {
return data ?? new Uint8Array(0);
}
+// Server-streaming gRPC-web call: onMessage fires per data frame as it
+// arrives; resolves when the stream ends (throws GrpcError on bad status).
+export async function grpcStream(base, method, reqBytes, { onMessage, signal }) {
+ const frame = new Uint8Array(5 + reqBytes.length);
+ new DataView(frame.buffer).setUint32(1, reqBytes.length);
+ frame.set(reqBytes, 5);
+ const resp = await fetch(`${base}/${method}`, {
+ method: 'POST',
+ headers: { 'content-type': 'application/grpc-web+proto', 'x-grpc-web': '1', 'pver': String(PVER) },
+ body: frame,
+ signal,
+ });
+ if (!resp.ok) throw new Error(`${method}: http ${resp.status}`);
+ const reader = resp.body.getReader();
+ let buf = new Uint8Array(0);
+ for (;;) {
+ const { done, value } = await reader.read();
+ if (done) break;
+ buf = concatBytes(buf, value);
+ for (;;) {
+ if (buf.length < 5) break;
+ const len = new DataView(buf.buffer, buf.byteOffset + 1, 4).getUint32(0);
+ if (buf.length < 5 + len) break;
+ const flag = buf[0];
+ const payload = buf.slice(5, 5 + len);
+ buf = buf.slice(5 + len);
+ if (flag & 0x80) {
+ const trailers = {};
+ for (const line of td.decode(payload).trim().split('\r\n')) {
+ const c = line.indexOf(':');
+ if (c > 0) trailers[line.slice(0, c).trim()] = line.slice(c + 1).trim();
+ }
+ const status = Number(trailers['grpc-status'] ?? 0);
+ if (status !== 0) throw new GrpcError(status, decodeURIComponent(trailers['grpc-message'] || ''), method);
+ } else {
+ onMessage(payload);
+ }
+ }
+ }
+}
+
// ---------------------------------------------------------------------------
// byte reader + bitcoin primitives (CompactSize, OutPoint, TxOut)
// ---------------------------------------------------------------------------
@@ -379,30 +420,39 @@ export async function handshake(ark, version = 'hal-ark-spike/0.0.1') {
// Read the mailbox; returns { messages: [{checkpoint, vtxos: [decoded]}], haveMore }
export async function readMailbox(ark, mailbox, checkpoint = 0) {
- const expiry = Math.floor(Date.now() / 1000) + 60;
- const auth = mailboxAuthorization(mailbox.privkey, mailbox.pubkey, expiry);
- const w = pbWriter();
- w.bytesField(1, mailbox.pubkey);
- w.bytesField(2, auth);
- if (checkpoint) w.varintField(3, checkpoint);
- const data = await grpcCall(ark, 'mailbox_server.MailboxService/ReadMailbox', w.finish());
+ const data = await grpcCall(ark, 'mailbox_server.MailboxService/ReadMailbox', mailboxRequestBytes(mailbox, checkpoint));
const messages = [];
let haveMore = false;
for (const { field, value } of pbFields(data)) {
if (field === 2) haveMore = Number(value) !== 0;
- if (field !== 1) continue;
- const msg = { checkpoint: 0, vtxos: [], kind: 'other' };
- for (const f of pbFields(value)) {
- if (f.field === 2) msg.checkpoint = Number(f.value);
- if (f.field === 1) { // ArkoorMessage
- msg.kind = 'arkoor';
- for (const a of pbFields(f.value)) {
- if (a.field === 1) msg.vtxos.push(decodeVtxo(a.value));
- }
+ if (field === 1) messages.push(decodeMailboxMessage(value));
+ }
+ return { messages, haveMore };
+}
+
+// Decode one MailboxMessage (also the unit of the SubscribeMailbox stream).
+export function decodeMailboxMessage(value) {
+ const msg = { checkpoint: 0, vtxos: [], kind: 'other' };
+ for (const f of pbFields(value)) {
+ if (f.field === 2) msg.checkpoint = Number(f.value);
+ if (f.field === 1) { // ArkoorMessage
+ msg.kind = 'arkoor';
+ for (const a of pbFields(f.value)) {
+ if (a.field === 1) msg.vtxos.push(decodeVtxo(a.value));
}
}
- messages.push(msg);
}
- return { messages, haveMore };
+ return msg;
+}
+
+// Build an authenticated MailboxRequest (shared by read + subscribe).
+export function mailboxRequestBytes(mailbox, checkpoint = 0) {
+ const expiry = Math.floor(Date.now() / 1000) + 60;
+ const auth = mailboxAuthorization(mailbox.privkey, mailbox.pubkey, expiry);
+ const w = pbWriter();
+ w.bytesField(1, mailbox.pubkey);
+ w.bytesField(2, auth);
+ if (checkpoint) w.varintField(3, checkpoint);
+ return w.finish();
}
generated by cgit v1.2.3 (git 2.51.0) at 2026-10-15 09:09:29 +0000