spillMessage function

SpillResult spillMessage(IpcMessage message, BlobBackend backend, int threshold)

Spill large payloads across an IpcMessage's value/state sites: Snapshot node states, Delta CellSet/SlotValue payloads + NodeAdd states, and CrdtSync op states. Returns a SpillResult with the rewritten message and the total bytes spilled.

Each inline payload above threshold is written to backend and replaced with a descriptor — the message stays small on the wire. Sites already carrying a descriptor are left untouched.

Implementation

SpillResult spillMessage(
    IpcMessage message, BlobBackend backend, int threshold) {
  var total = 0;

  if (message is IpcMessageSnapshot) {
    final snap = message.value;
    final nodes = <NodeSnapshot>[];
    for (final node in snap.nodes) {
      final (state, spilled) = _spillState(node.state, backend, threshold);
      total += spilled;
      nodes.add(NodeSnapshot(node.node, node.typeTag, state, key: node.key));
    }
    return SpillResult(
      IpcMessageSnapshot(Snapshot(
        epoch: snap.epoch,
        nodes: nodes,
        edges: snap.edges,
        roots: snap.roots,
      )),
      total,
    );
  }

  if (message is IpcMessageDelta) {
    final delta = message.value;
    final ops = <DeltaOp>[];
    for (final op in delta.ops) {
      if (op is DeltaOpCellSet) {
        final (payload, spilled) = spillValue(op.payload, backend, threshold);
        total += spilled;
        ops.add(DeltaOpCellSet(op.node, payload));
      } else if (op is DeltaOpSlotValue) {
        final (payload, spilled) = spillValue(op.payload, backend, threshold);
        total += spilled;
        ops.add(DeltaOpSlotValue(op.node, payload));
      } else if (op is DeltaOpNodeAdd) {
        final (state, spilled) = _spillState(op.state, backend, threshold);
        total += spilled;
        ops.add(DeltaOpNodeAdd(op.node, op.typeTag, state, key: op.key));
      } else {
        ops.add(op);
      }
    }
    return SpillResult(
      IpcMessageDelta(
          Delta(baseEpoch: delta.baseEpoch, epoch: delta.epoch, ops: ops)),
      total,
    );
  }

  if (message is IpcMessageCrdtSync) {
    final sync = message.value;
    final ops = <CrdtOp>[];
    for (final op in sync.ops) {
      final (state, spilled) = spillValue(op.state, backend, threshold);
      total += spilled;
      ops.add(CrdtOp(node: op.node, stamp: op.stamp, state: state, key: op.key));
    }
    return SpillResult(
      IpcMessageCrdtSync(CrdtSync(frontier: sync.frontier, ops: ops)),
      total,
    );
  }

  return SpillResult(message, 0);
}