Merge pull request #559 from OriginTrail/feat/publisher-storage-ack-typed-declines

feat(publisher): structured StorageACK declines instead of stream resets

Author: branarakic

packages/core/src/proto/index.ts

@@ -87,8 +87,13 @@ export {
 
 export {
   type StorageACKMsg,
+  type StorageACKDeclineCode,
+  STORAGE_ACK_DECLINE_CODES,
+  TRANSIENT_STORAGE_ACK_DECLINE_CODES,
   encodeStorageACK,
   decodeStorageACK,
+  isStorageACKDecline,
+  isTransientStorageACKDeclineCode,
 } from './storage-ack.js';
 
 export {

packages/core/src/proto/storage-ack.ts

@@ -26,12 +26,88 @@ const { Type, Field } = protobuf;
  * byte-for-byte.
  */
 
+/**
+ * Declinable reasons a core node can return on `/dkg/10.0.0/storage-ack`
+ * instead of a signed ACK. These are situations where the core
+ * legitimately cannot produce an ACK for THIS PEER right now — a
+ * well-formed publish request that this specific core just can't
+ * satisfy.
+ *
+ * Codes split into two classes that the publisher treats differently:
+ *
+ * - **Transient** ({@link TRANSIENT_STORAGE_ACK_DECLINE_CODES}):
+ *   the local condition is expected to clear on its own (e.g. SWM is
+ *   catching up via gossip). The publisher retries the same peer with
+ *   the existing transport backoff before giving up — keeps quorum
+ *   reachable when replication is just slightly behind the publish.
+ *
+ * - **Permanent** (every other code):
+ *   the condition will not change on a fast retry (e.g. the operational
+ *   signer was rotated off-chain). The publisher deselects this peer
+ *   for THIS request and moves on to others; the reason is surfaced
+ *   in the final error if quorum still fails.
+ *
+ * Malformed-request errors are NOT declines and do NOT belong here.
+ * The handler keeps `throw`ing on those so the publisher sees them as
+ * stream errors with the original message and surfaces them to the
+ * caller immediately rather than fanning out to every core looking
+ * for a different answer.
+ *
+ * String values are part of the wire format: they are populated into
+ * `StorageACK.declineCode` and surfaced in publisher logs / error
+ * messages. Keep them stable across releases. Adding a new code is a
+ * non-breaking change (older publishers see it as the catch-all
+ * "unknown decline" path); renaming or removing one IS breaking.
+ */
+export const STORAGE_ACK_DECLINE_CODES = {
+  /** SWM CONSTRUCT returned no quads for the request. */
+  NO_DATA_IN_SWM: 'NO_DATA_IN_SWM',
+  /** SWM has data but its merkle root does not match the publisher's. */
+  MERKLE_MISMATCH_IN_SWM: 'MERKLE_MISMATCH_IN_SWM',
+  /** Operational signer was just removed / rotated off-chain. */
+  SIGNER_NOT_REGISTERED: 'SIGNER_NOT_REGISTERED',
+} as const;
+
+export type StorageACKDeclineCode =
+  (typeof STORAGE_ACK_DECLINE_CODES)[keyof typeof STORAGE_ACK_DECLINE_CODES];
+
+/**
+ * Decline codes whose root cause is expected to clear on its own
+ * (typically SWM replication catching up via gossip). The publisher
+ * retries these against the same peer through the normal transport
+ * backoff before giving up, so a peer that would have ACKed seconds
+ * later still contributes to quorum.
+ *
+ * Membership of this set is part of the protocol contract between
+ * publisher and core — promoting / demoting a code is a behavior
+ * change, not a wire change.
+ */
+export const TRANSIENT_STORAGE_ACK_DECLINE_CODES: ReadonlySet<string> = new Set<string>([
+  STORAGE_ACK_DECLINE_CODES.NO_DATA_IN_SWM,
+  STORAGE_ACK_DECLINE_CODES.MERKLE_MISMATCH_IN_SWM,
+]);
+
+/** True iff `code` names a decline the publisher should retry rather than treat as permanent. */
+export function isTransientStorageACKDeclineCode(code: string | undefined): boolean {
+  return typeof code === 'string' && TRANSIENT_STORAGE_ACK_DECLINE_CODES.has(code);
+}
+
+/**
+ * Wire schema. Fields 1–5 are the original ACK shape; fields 6–7 carry
+ * a decline payload. Two optional strings (rather than a `oneof`) keep
+ * the change strictly additive: old encoders never set the new fields,
+ * old decoders silently ignore them, so cross-version traffic is
+ * unchanged. New decoders inspect `declineCode` first — when it is
+ * non-empty the message is a decline and the ACK fields are unset.
+ */
 export const StorageACKSchema = new Type('StorageACK')
   .add(new Field('merkleRoot', 1, 'bytes'))
   .add(new Field('coreNodeSignatureR', 2, 'bytes'))
   .add(new Field('coreNodeSignatureVS', 3, 'bytes'))
   .add(new Field('contextGraphId', 4, 'string'))
-  .add(new Field('nodeIdentityId', 5, 'uint64'));
+  .add(new Field('nodeIdentityId', 5, 'uint64'))
+  .add(new Field('declineCode', 6, 'string'))
+  .add(new Field('declineMessage', 7, 'string'));
 
 type Long = { low: number; high: number; unsigned: boolean };
 
@@ -41,6 +117,31 @@ export interface StorageACKMsg {
   coreNodeSignatureVS: Uint8Array;
   contextGraphId: string;
   nodeIdentityId: number | Long;
+  /**
+   * When non-empty, this message is a decline rather than a signed ACK
+   * — see {@link STORAGE_ACK_DECLINE_CODES}. Old senders never set this
+   * field; old receivers ignore it. New receivers MUST check this
+   * before treating the message as an ACK (signature/merkleRoot are
+   * unset on declines).
+   */
+  declineCode?: string;
+  /**
+   * Human-readable reason that accompanies `declineCode`. Surfaced into
+   * publisher logs and the final `storage_ack_insufficient` error so
+   * operators can diagnose hosting / replication issues without having
+   * to ssh into individual cores.
+   */
+  declineMessage?: string;
+}
+
+/**
+ * Convenience: returns true iff the message is a decline (i.e.
+ * `declineCode` is set to a non-empty string). Keeps callers from
+ * having to remember the empty-string-as-undefined idiom that
+ * protobufjs uses for unset string fields.
+ */
+export function isStorageACKDecline(msg: StorageACKMsg): boolean {
+  return typeof msg.declineCode === 'string' && msg.declineCode.length > 0;
 }
 
 export function encodeStorageACK(msg: StorageACKMsg): Uint8Array {

packages/core/test/v10-proto.test.ts

@@ -130,6 +130,61 @@ describe('StorageACKMsg', () => {
   it('deterministic encoding', () => {
     expect(encodeStorageACK(ack)).toEqual(encodeStorageACK(ack));
   });
+
+  it('decodes an old ACK (no decline fields) without populating declineCode', async () => {
+    const decoded = decodeStorageACK(encodeStorageACK(ack));
+    expect(decoded.declineCode == null || decoded.declineCode === '').toBe(true);
+    expect(decoded.declineMessage == null || decoded.declineMessage === '').toBe(true);
+    const { isStorageACKDecline } = await import('../src/proto/storage-ack.js');
+    expect(isStorageACKDecline(decoded)).toBe(false);
+  });
+
+  it('decline-only message: empty ACK fields + populated decline code/message round-trip', async () => {
+    const { STORAGE_ACK_DECLINE_CODES, isStorageACKDecline } = await import('../src/proto/storage-ack.js');
+    const decline: StorageACKMsg = {
+      merkleRoot: new Uint8Array(0),
+      coreNodeSignatureR: new Uint8Array(0),
+      coreNodeSignatureVS: new Uint8Array(0),
+      contextGraphId: '15',
+      nodeIdentityId: 0,
+      declineCode: STORAGE_ACK_DECLINE_CODES.NO_DATA_IN_SWM,
+      declineMessage:
+        'No data found in SWM graph did:dkg:context-graph:15/_shared_memory for entities: urn:a, urn:b',
+    };
+    const decoded = decodeStorageACK(encodeStorageACK(decline));
+    expect(decoded.declineCode).toBe('NO_DATA_IN_SWM');
+    expect(decoded.declineMessage).toContain('No data found in SWM graph');
+    expect(decoded.contextGraphId).toBe('15');
+    expect(isStorageACKDecline(decoded)).toBe(true);
+    expect(new Uint8Array(decoded.merkleRoot).length).toBe(0);
+    expect(new Uint8Array(decoded.coreNodeSignatureR).length).toBe(0);
+  });
+
+  it('a new decoder reading bytes from an old encoder still yields a valid ACK (forward compat)', () => {
+    // Literal pre-decline wire shape from the old 5-field schema:
+    // 1=merkleRoot, 2=signatureR, 3=signatureVS, 4=contextGraphId,
+    // 5=nodeIdentityId. Keeping this as bytes catches regressions where
+    // the new schema stops decoding historical ACK payloads even though
+    // the current encoder still omits unset decline fields.
+    const wire = Uint8Array.from([
+      0x0a, 0x20,
+      ...new Array(32).fill(0xa5),
+      0x12, 0x20,
+      ...new Array(32).fill(0x11),
+      0x1a, 0x20,
+      ...new Array(32).fill(0x22),
+      0x22, 0x06,
+      0x63, 0x67, 0x2d, 0x31, 0x30, 0x30,
+      0x28, 0x07,
+    ]);
+    const decoded = decodeStorageACK(wire);
+    expect(decoded.contextGraphId).toBe('cg-100');
+    expect(Number(decoded.nodeIdentityId)).toBe(7);
+    expect(new Uint8Array(decoded.merkleRoot)).toEqual(new Uint8Array(32).fill(0xa5));
+    expect(new Uint8Array(decoded.coreNodeSignatureR)).toEqual(new Uint8Array(32).fill(0x11));
+    expect(new Uint8Array(decoded.coreNodeSignatureVS)).toEqual(new Uint8Array(32).fill(0x22));
+    expect(decoded.declineCode == null || decoded.declineCode === '').toBe(true);
+  });
 });
 
 // ── SwmShareAck (rc.9 PR-D) ───────────────────────────────────────────