macula_frame (macula v7.0.0)
View SourceCBOR-encoded wire frames for Macula V2 (Part 6 §3 canonical wire).
A wire frame is a length-prefixed deterministic CBOR map:
<<Length:32/big, Cbor/binary>>
where Cbor is the RFC 8949 §4.2.1 deterministic encoding of a single map. The map carries the common header fields (Part 6 §3) plus type-specific fields and an Ed25519 signature.
PLAN_WIRE_CBOR.md migrated this codec from BERT to CBOR so hecate-station and the macula 3.x SDK share a wire format. Frame schemas (atom-keyed maps in process memory) are unchanged; the codec transparently round-trips atom keys/values to text strings on the wire and reconstitutes them via binary_to_existing_atom on decode.
Phase 1 covers CONNECT / HELLO / GOODBYE. Phase 2 adds SWIM. Phase 3 (Session 3.4) adds the DHT operation frames from Part 6 §7: PING / PONG, FIND_NODE / NODES, FIND_VALUE / VALUE, STORE / STORE_ACK, REPLICATE / REPLICATE_ACK. Phase 4 (Session 4.1) adds the CALL / RESULT / ERROR frames from Part 6 §5 plus the BOLT#4 error taxonomy in hecate_bolt4. PUBLISH frames land later.
Signatures are Ed25519 over "macula-v2-frame\0" ++ canonical_cbor(unsigned) where canonical_cbor is macula_record_cbor:encode/1 (RFC 8949 §4.2.1 deterministic).
Summary
Functions
Is this whole frame sendable? Used by macula_peering:send_frame/2, which is the single seam every producer passes through.
Is this term admissible as a frame payload?
Decode a single length-prefixed frame from the head of a buffer. Returns {ok, Frame, RestBuffer}, {more, BytesNeeded} if the buffer is short, or {error, Reason} if the framing is malformed.
Render a rejection as a sentence, with the remedy where there is one. The operator reading a log at 03:00 is not reading edoc.
Drain all complete frames from a buffer. Returns the list of frames (in order) and the remaining (incomplete) buffer.
Add publisher_sig to a PUBLISH or EVENT frame: the publisher's Ed25519 signature over (topic, realm, publisher, seq, payload). Identity must be the key pair / private key of the pubkey in the frame's publisher field.
Verify a frame's publisher_sig against its publisher field. Returns {ok, Frame} on success, {error, Reason} otherwise (no_publisher_sig when the field is absent — callers that want it MUST treat absence as a verification failure, not as "trusted").
Types
-type advertise_spec() :: #{realm := id256(), procedure := binary(), advertiser := macula_identity:pubkey(), options => map()}.
-type call_error_spec() :: #{call_id := call_id(), code := macula_bolt4:code(), reported_by := macula_identity:pubkey(), detail => binary() | undefined, offending_hop => macula_identity:pubkey() | undefined, source_route_partial => binary()}.
-type call_id() :: <<_:128>>.
-type call_spec() :: #{call_id := call_id(), procedure := binary(), realm := id256(), payload := term(), deadline_ms := integer(), caller := macula_identity:pubkey(), source_route => binary(), retry_budget => non_neg_integer()}.
-type cancel_spec() :: #{blocks := [mcid()]}.
-type connect_spec() :: #{node_id := macula_identity:pubkey(), station_id := macula_identity:pubkey(), realms := [macula_identity:pubkey()], capabilities := non_neg_integer(), puzzle_evidence := <<_:256>>, addresses => [map()], site => map() | undefined, endorsements => [map()]}.
-type country() :: <<_:16>>.
-type delivery_channel() :: plumtree | dht | direct.
-type event_spec() :: #{topic := binary(), realm := id256(), publisher := macula_identity:pubkey(), seq := non_neg_integer(), payload := term(), delivered_via := delivery_channel(), publisher_sig => binary()}.
-type find_node_spec() :: #{key := id256(), origin := macula_identity:pubkey(), depth := non_neg_integer()}.
-type find_value_spec() :: #{key := id256(), origin := macula_identity:pubkey()}.
-type frame() :: map().
-type frame_type() ::
connect | hello | goodbye | swim_ping | swim_ack | swim_suspect | swim_confirm | ping | pong |
find_node | nodes | find_value | value | store | store_ack | replicate | replicate_ack |
call | result | error | hyparview_join | hyparview_forward_join | hyparview_neighbor |
hyparview_disconnect | hyparview_shuffle | hyparview_shuffle_reply | plumtree_gossip |
plumtree_ihave | plumtree_graft | plumtree_prune | publish | subscribe | unsubscribe | event |
advertise | unadvertise | stream_open | stream_data | stream_end | stream_error |
stream_reply | want | have | block | manifest_req | manifest_res | cancel.
-type have_entry() :: #{mcid := mcid(), size := non_neg_integer()}.
-type have_spec() :: #{blocks := [have_entry()]}.
-type hello_spec() :: #{node_id := macula_identity:pubkey(), station_id := macula_identity:pubkey(), realms := [macula_identity:pubkey()], capabilities := non_neg_integer(), accepted := boolean(), negotiated_capabilities := non_neg_integer(), addresses => [map()], site => map() | undefined, refusal_code => non_neg_integer() | undefined}.
-type hyparview_disconnect_spec() :: #{realm := id256()}.
-type hyparview_forward_join_spec() :: #{realm := id256(), new_member := macula_identity:pubkey(), ttl := non_neg_integer(), arwl := non_neg_integer(), prwl := non_neg_integer()}.
-type hyparview_join_spec() :: #{realm := id256(), new_member := macula_identity:pubkey()}.
-type hyparview_neighbor_spec() :: #{realm := id256(), priority := neighbor_priority()}.
-type hyparview_shuffle_reply_spec() :: #{realm := id256(), peer_sample := [macula_identity:pubkey()]}.
-type hyparview_shuffle_spec() :: #{realm := id256(), origin := macula_identity:pubkey(), ttl := non_neg_integer(), peer_sample := [macula_identity:pubkey()]}.
-type id256() :: <<_:256>>.
-type manifest_req_spec() :: #{mcid := mcid()}.
-type mcid() :: <<_:272>>.
-type member_state() :: alive | suspect | confirmed_failed.
-type msg_id() :: <<_:128>>.
-type neighbor_priority() :: high | low.
-type nodes_spec() :: #{key := id256(), nodes := [station_ref()]}.
-type nonce128() :: <<_:128>>.
-type ping_spec() :: #{nonce := nonce128()}.
-type plumtree_gossip_spec() :: #{realm := id256(), msg_id := msg_id(), round := non_neg_integer(), payload := term()}.
-type plumtree_graft_spec() :: #{realm := id256(), msg_id := msg_id(), round := non_neg_integer()}.
-type plumtree_ihave_spec() :: #{realm := id256(), msg_id := msg_id(), round := non_neg_integer()}.
-type plumtree_prune_spec() :: #{realm := id256()}.
-type pong_spec() :: #{nonce := nonce128()}.
-type publish_spec() :: #{topic := binary(), realm := id256(), publisher := macula_identity:pubkey(), seq := non_neg_integer(), payload := term(), published_at_ms := non_neg_integer(), ttl_ms => non_neg_integer() | undefined, publisher_sig => binary()}.
-type replicate_spec() :: #{record := macula_record:record(), new_custodian := boolean()}.
-type result_spec() :: #{call_id := call_id(), payload := term(), responded_by := macula_identity:pubkey(), source_route_reverse => binary()}.
-type station_ref() :: #{node_id := macula_identity:pubkey(), station_id := macula_identity:pubkey(), addresses := [map()], tier := tier(), asn := non_neg_integer() | undefined, country := country(), last_seen_at := pos_integer()}.
-type station_ref_spec() :: #{node_id := macula_identity:pubkey(), station_id := macula_identity:pubkey(), addresses => [map()], tier := tier(), asn => non_neg_integer() | undefined, country := country(), last_seen_at := pos_integer()}.
-type store_spec() :: #{record := macula_record:record()}.
-type stream_data_spec() :: #{stream_id := stream_id(), seq := non_neg_integer(), encoding := stream_encoding(), body := term()}.
-type stream_encoding() :: raw | msgpack.
-type stream_end_spec() :: #{stream_id := stream_id(), role := stream_role()}.
-type stream_id() :: <<_:128>>.
-type stream_mode() :: server_stream | client_stream | bidi.
-type stream_open_spec() :: #{stream_id := stream_id(), procedure := binary(), realm := id256(), mode := stream_mode(), args := term(), deadline_ms := integer(), caller := macula_identity:pubkey(), source_route => binary(), retry_budget => non_neg_integer()}.
-type stream_reply_spec() :: #{stream_id := stream_id(), payload := term(), responded_by := macula_identity:pubkey()}.
-type stream_role() :: send | both.
-type subscribe_spec() :: #{topic := binary(), realm := id256(), subscriber := macula_identity:pubkey(), filter => term() | undefined, options => map()}.
-type swim_ack_spec() :: #{round := non_neg_integer(), responder := macula_identity:pubkey(), incarnation := non_neg_integer(), piggyback => [swim_update()]}.
-type swim_ping_spec() :: #{round := non_neg_integer(), incarnation := non_neg_integer(), piggyback => [swim_update()]}.
-type swim_suspect_spec() :: #{target := macula_identity:pubkey(), target_incarnation := non_neg_integer(), suspected_by := macula_identity:pubkey(), ttl := non_neg_integer()}.
-type swim_update() :: #{target := macula_identity:pubkey(), state := member_state(), incarnation := non_neg_integer(), observed_at := pos_integer(), by := macula_identity:pubkey(), signature => <<_:512>>}.
-type swim_update_spec() :: #{target := macula_identity:pubkey(), state := member_state(), incarnation := non_neg_integer(), observed_at := pos_integer(), by := macula_identity:pubkey()}.
-type tier() :: 0..4.
-type unadvertise_spec() :: #{realm := id256(), procedure := binary(), advertiser := macula_identity:pubkey()}.
-type unsubscribe_spec() :: #{topic := binary(), realm := id256(), subscriber := macula_identity:pubkey()}.
-type value_spec() :: #{key := id256(), records := [macula_record:record()]}.
-type want_entry() :: #{mcid := mcid(), priority => want_priority()}.
-type want_priority() :: 0..255.
-type want_spec() :: #{blocks := [want_entry()]}.
Functions
-spec advertise(advertise_spec()) -> frame().
-spec block(block_spec()) -> frame().
-spec call_error(call_error_spec()) -> frame().
-spec cancel(cancel_spec()) -> frame().
Is this whole frame sendable? Used by macula_peering:send_frame/2, which is the single seam every producer passes through.
record / records are excluded: prepare_records/1 hands those to macula_record:encode/1, a different encoder with its own rules, so judging them by payload rules would reject frames that encode fine.
KNOWN LIMIT, stated rather than papered over. That exclusion means record-bearing frames (STORE, REPLICATE, VALUE) get weaker guarantees than payload-bearing ones: an unsigned record passes here and then fails in macula_record:encode/1, and byte_floor/1 never sees record bytes, so an oversized record is not caught synchronously. Neither can kill the connection — encode_or_drop/2 catches both — but the caller is told ok and the frame is dropped. So "callers get a structured reason" is true for payload-bearing frames and NOT for record-bearing ones. Closing it means teaching this function the record shape and folding record size into the floor.
Is this term admissible as a frame payload?
Returns ok, or {error, {unsupported_payload_type, Type, Path}} where Path locates the offending value inside the term (map keys and zero-based list indices, outermost first).
WHY THIS EXISTS. macula_peering:send_frame/2 is a cast, so the frame is encoded later, in the shared peering connection process, with no try/catch around it. A term the codec cannot represent therefore does not fail the publisher — it kills the connection, taking every other producer's in-flight traffic with it, while the publisher was told ok. Checking here, in the caller's process before the cast, is what makes that ok mean something.
This function must agree exactly with to_wire/1 followed by macula_record_cbor:encode/1. That is why it lives beside them rather than in a validation module: the two cannot drift apart without the agreement property test in macula_frame_tests going red.
FLOATS ARE CARRIED, as of the float support in macula_record_cbor. They used to be rejected here, and before that to_wire/1 silently rewrote them as six-decimal text. Both were workarounds for the canonical encoder lacking a float clause, which was never a CBOR limitation: RFC 8949 major type 7 is floats. The encoder now emits binary64, so callers no longer scale to integers to get a number across.
-spec connect(connect_spec()) -> frame().
-spec decode(binary()) -> {ok, frame(), binary()} | {more, pos_integer()} | {error, term()}.
Decode a single length-prefixed frame from the head of a buffer. Returns {ok, Frame, RestBuffer}, {more, BytesNeeded} if the buffer is short, or {error, Reason} if the framing is malformed.
-spec event(event_spec()) -> frame().
-spec explain(term()) -> unicode:chardata().
Render a rejection as a sentence, with the remedy where there is one. The operator reading a log at 03:00 is not reading edoc.
-spec find_node(find_node_spec()) -> frame().
-spec find_value(find_value_spec()) -> frame().
-spec goodbye(atom(), binary() | undefined, non_neg_integer()) -> frame().
-spec hello(hello_spec()) -> frame().
-spec hyparview_disconnect(hyparview_disconnect_spec()) -> frame().
-spec hyparview_forward_join(hyparview_forward_join_spec()) -> frame().
-spec hyparview_join(hyparview_join_spec()) -> frame().
-spec hyparview_neighbor(hyparview_neighbor_spec()) -> frame().
-spec hyparview_shuffle(hyparview_shuffle_spec()) -> frame().
-spec hyparview_shuffle_reply(hyparview_shuffle_reply_spec()) -> frame().
-spec manifest_req(manifest_req_spec()) -> frame().
-spec manifest_res(manifest_res_spec()) -> frame().
-spec nodes(nodes_spec()) -> frame().
Drain all complete frames from a buffer. Returns the list of frames (in order) and the remaining (incomplete) buffer.
-spec plumtree_gossip(plumtree_gossip_spec()) -> frame().
-spec plumtree_graft(plumtree_graft_spec()) -> frame().
-spec plumtree_ihave(plumtree_ihave_spec()) -> frame().
-spec plumtree_prune(plumtree_prune_spec()) -> frame().
-spec publish(publish_spec()) -> frame().
-spec replicate(replicate_spec()) -> frame().
-spec replicate_ack(replicate_ack_spec()) -> frame().
-spec result(result_spec()) -> frame().
-spec sign(frame(), macula_identity:key_pair() | macula_identity:privkey()) -> frame().
-spec sign_publisher(frame(), macula_identity:key_pair() | macula_identity:privkey()) -> frame().
Add publisher_sig to a PUBLISH or EVENT frame: the publisher's Ed25519 signature over (topic, realm, publisher, seq, payload). Identity must be the key pair / private key of the pubkey in the frame's publisher field.
-spec sign_swim_update(swim_update(), macula_identity:key_pair() | macula_identity:privkey()) -> swim_update().
-spec station_ref(station_ref_spec()) -> station_ref().
-spec store(store_spec()) -> frame().
-spec store_ack(store_ack_spec()) -> frame().
-spec stream_data(stream_data_spec()) -> frame().
-spec stream_end(stream_end_spec()) -> frame().
-spec stream_error(stream_error_spec()) -> frame().
-spec stream_open(stream_open_spec()) -> frame().
-spec stream_reply(stream_reply_spec()) -> frame().
-spec subscribe(subscribe_spec()) -> frame().
-spec swim_ack(swim_ack_spec()) -> frame().
-spec swim_confirm(swim_suspect_spec()) -> frame().
-spec swim_ping(swim_ping_spec()) -> frame().
-spec swim_suspect(swim_suspect_spec()) -> frame().
-spec swim_update(swim_update_spec()) -> swim_update().
-spec unadvertise(unadvertise_spec()) -> frame().
-spec unsubscribe(unsubscribe_spec()) -> frame().
-spec value(value_spec()) -> frame().
-spec verify(frame(), macula_identity:pubkey()) -> {ok, frame()} | {error, term()}.
Verify a frame's publisher_sig against its publisher field. Returns {ok, Frame} on success, {error, Reason} otherwise (no_publisher_sig when the field is absent — callers that want it MUST treat absence as a verification failure, not as "trusted").
-spec verify_swim_update(swim_update()) -> {ok, swim_update()} | {error, term()}.