zap: carry the cross-chain atomic seam across the plugin boundary

One atomic.Memory exists per node and every chain gets a handle carved out
of it. A VM linked into the node reaches its handle off runtime.Runtime; a
VM hosted as a ZAP plugin — the C-Chain EVM and the dexvm both are — does
not, because the node rebuilds the plugin's Runtime from InitializeRequest
fields and neither an interface over a live database (SharedMemory) nor the
chain-alias lookup (BCLookup) survives a struct literal. The plugin's
SharedMemory was therefore nil and every DEX 0x9999 settlement reverted
"requires the cross-chain atomic capability".

Adds the wire for it: MsgAtomicGet/MsgAtomicApply, and AtomicServerAddr +
DChainID appended to InitializeRequest. The node binds a ZAP server over
its own handle and the plugin dials it, so the node is the server and the
existing strictly request/response transport carries this with no duplex
framing and no gRPC — the same shape already used for DBServerAddr.

Indexed is deliberately not proxied: no settlement path calls it.

Both skew directions are pinned by test rather than reasoned about, since a
wire change that forks a mixed fleet is the failure this design exists to
avoid. New node -> old plugin: the appended bytes are inert, the plugin
keeps its nil handle and its existing fail-closed revert. Old node -> new
plugin: Remaining() is 0, so the capability reads as ABSENT instead of
raising the ErrUnexpectedEOF that the Capabilities append once caused.
Neither direction changes how any block executes.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
2026-08-06 14:08:52 -07:00
co-authored by Hanzo Dev
parent 7a86664099
commit 8743bbe098
4 changed files with 538 additions and 3 deletions
+199
View File
@@ -0,0 +1,199 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package zap
// atomic.go is the wire for the cross-chain atomic shared-memory surface — the
// primary-network import/export primitive platformvm, the dexvm and the DEX
// 0x9999 settlement seam all move value through.
//
// WHY IT IS ON THE WIRE AT ALL. One atomic.Memory exists per NODE; every chain
// gets a per-chain SharedMemory handle carved out of it. A VM linked into the
// node reaches its handle straight off runtime.Runtime. A VM hosted as a ZAP
// plugin — the C-Chain EVM and the dexvm both are — does not: the node rebuilds
// the plugin's Runtime from the InitializeRequest fields, and neither the handle
// (an interface over a live database) nor BCLookup can be copied into a struct
// literal. So the plugin's Runtime.SharedMemory was nil and every settlement
// reverted "requires the cross-chain atomic capability".
//
// DIRECTION AND FRAMING. The node binds a per-chain ZAP server over its own
// SharedMemory handle and names it in InitializeRequest.AtomicServerAddr; the
// plugin dials that addr as an ordinary ZAP client. The node is the SERVER, so
// the existing strictly request/response transport carries this unchanged — no
// duplex framing, no second demultiplexer, no gRPC. It is the identical shape
// the node already uses for DBServerAddr.
//
// TYPES. This package stays dependency-free on purpose (luxfi/vm imports
// luxfi/api, so the reverse would be a cycle). IDs are plain 32-byte slices and
// the request set is flattened; the plugin side converts to and from
// atomic.Requests.
// AtomicGetRequest asks the node for the values a peer chain published for this
// chain. It is SharedMemory.Get(peerChainID, keys) verbatim.
type AtomicGetRequest struct {
PeerChainID []byte
Keys [][]byte
}
// Encode serializes AtomicGetRequest to the buffer
func (m *AtomicGetRequest) Encode(buf *Buffer) {
buf.WriteBytes(m.PeerChainID)
buf.WriteUint32(uint32(len(m.Keys)))
for _, k := range m.Keys {
buf.WriteBytes(k)
}
}
// Decode deserializes AtomicGetRequest from the reader
func (m *AtomicGetRequest) Decode(r *Reader) error {
var err error
if m.PeerChainID, err = r.ReadBytes(); err != nil {
return err
}
count, err := r.ReadCount()
if err != nil {
return err
}
m.Keys = make([][]byte, count)
for i := uint32(0); i < count; i++ {
if m.Keys[i], err = r.ReadBytes(); err != nil {
return err
}
}
return nil
}
// AtomicGetResponse carries the values, positionally aligned with the request's
// keys. SharedMemory.Get guarantees len(values) == len(keys); an absent key is
// an EMPTY value at its index, never a short slice — the caller distinguishes
// "no such object" from "object present" by length, so that invariant is part of
// the wire contract and is asserted on decode by the caller.
type AtomicGetResponse struct {
Values [][]byte
}
// Encode serializes AtomicGetResponse to the buffer
func (m *AtomicGetResponse) Encode(buf *Buffer) {
buf.WriteUint32(uint32(len(m.Values)))
for _, v := range m.Values {
buf.WriteBytes(v)
}
}
// Decode deserializes AtomicGetResponse from the reader
func (m *AtomicGetResponse) Decode(r *Reader) error {
count, err := r.ReadCount()
if err != nil {
return err
}
m.Values = make([][]byte, count)
for i := uint32(0); i < count; i++ {
if m.Values[i], err = r.ReadBytes(); err != nil {
return err
}
}
return nil
}
// AtomicElement is one published cross-chain object: the value stored under Key,
// plus the traits it is indexed by. Mirrors atomic.Element field-for-field.
type AtomicElement struct {
Key []byte
Value []byte
Traits [][]byte
}
// AtomicChainRequests is one peer chain's operations. Mirrors atomic.Requests.
type AtomicChainRequests struct {
PeerChainID []byte
RemoveRequests [][]byte
PutRequests []*AtomicElement
}
// AtomicApplyRequest is SharedMemory.Apply(requests) — the atomic commit of a
// block's cross-chain puts and removes.
//
// NO BATCH CROSSES THIS WIRE. In-process, Apply takes the caller's database
// batch so the state commit and the shared-memory mutation land as one write; a
// batch is a live handle over a database the node owns and cannot be serialized.
// The plugin therefore relies on the property its flush window already has: the
// ops are derived from a monotone seq in CONSENSUS state, so a replayed window
// yields byte-identical Puts and Removes. Both are idempotent under replay, so
// at-least-once delivery of this message has exactly-once effect — which is why
// the caller must advance its flushed-seq marker only AFTER this call returns.
type AtomicApplyRequest struct {
Requests []*AtomicChainRequests
}
// Encode serializes AtomicApplyRequest to the buffer
func (m *AtomicApplyRequest) Encode(buf *Buffer) {
buf.WriteUint32(uint32(len(m.Requests)))
for _, req := range m.Requests {
buf.WriteBytes(req.PeerChainID)
buf.WriteUint32(uint32(len(req.RemoveRequests)))
for _, rm := range req.RemoveRequests {
buf.WriteBytes(rm)
}
buf.WriteUint32(uint32(len(req.PutRequests)))
for _, e := range req.PutRequests {
buf.WriteBytes(e.Key)
buf.WriteBytes(e.Value)
buf.WriteUint32(uint32(len(e.Traits)))
for _, tr := range e.Traits {
buf.WriteBytes(tr)
}
}
}
}
// Decode deserializes AtomicApplyRequest from the reader
func (m *AtomicApplyRequest) Decode(r *Reader) error {
chainCount, err := r.ReadCount()
if err != nil {
return err
}
m.Requests = make([]*AtomicChainRequests, chainCount)
for i := uint32(0); i < chainCount; i++ {
req := &AtomicChainRequests{}
if req.PeerChainID, err = r.ReadBytes(); err != nil {
return err
}
rmCount, err := r.ReadCount()
if err != nil {
return err
}
req.RemoveRequests = make([][]byte, rmCount)
for j := uint32(0); j < rmCount; j++ {
if req.RemoveRequests[j], err = r.ReadBytes(); err != nil {
return err
}
}
putCount, err := r.ReadCount()
if err != nil {
return err
}
req.PutRequests = make([]*AtomicElement, putCount)
for j := uint32(0); j < putCount; j++ {
e := &AtomicElement{}
if e.Key, err = r.ReadBytes(); err != nil {
return err
}
if e.Value, err = r.ReadBytes(); err != nil {
return err
}
trCount, err := r.ReadCount()
if err != nil {
return err
}
e.Traits = make([][]byte, trCount)
for k := uint32(0); k < trCount; k++ {
if e.Traits[k], err = r.ReadBytes(); err != nil {
return err
}
}
req.PutRequests[j] = e
}
m.Requests[i] = req
}
return nil
}
+260
View File
@@ -0,0 +1,260 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package zap
import (
"bytes"
"testing"
)
// roundTrip encodes m, hands the bytes to decode, and returns the payload so a
// test can also assert on framing.
func encodeToBytes(enc func(*Buffer)) []byte {
buf := GetBuffer()
defer PutBuffer(buf)
enc(buf)
out := make([]byte, len(buf.Bytes()))
copy(out, buf.Bytes())
return out
}
func TestAtomicGetRequest_RoundTrip(t *testing.T) {
in := &AtomicGetRequest{
PeerChainID: bytes.Repeat([]byte{0xD1}, 32),
Keys: [][]byte{bytes.Repeat([]byte{0x01}, 32), bytes.Repeat([]byte{0x02}, 32)},
}
payload := encodeToBytes(in.Encode)
var out AtomicGetRequest
if err := out.Decode(NewReader(payload)); err != nil {
t.Fatalf("decode: %v", err)
}
if !bytes.Equal(out.PeerChainID, in.PeerChainID) {
t.Fatalf("peerChainID = %x, want %x", out.PeerChainID, in.PeerChainID)
}
if len(out.Keys) != len(in.Keys) {
t.Fatalf("keys len = %d, want %d", len(out.Keys), len(in.Keys))
}
for i := range in.Keys {
if !bytes.Equal(out.Keys[i], in.Keys[i]) {
t.Fatalf("key[%d] = %x, want %x", i, out.Keys[i], in.Keys[i])
}
}
}
// TestAtomicGetResponse_AbsentIsEmptyNotShort pins the invariant the settlement
// path depends on: SharedMemory.Get returns a values slice the SAME length as
// keys, with an absent object represented as an EMPTY value at its index. If the
// wire collapsed empties the caller would mis-align values with keys and could
// bind a credit to the WRONG object.
func TestAtomicGetResponse_AbsentIsEmptyNotShort(t *testing.T) {
in := &AtomicGetResponse{
Values: [][]byte{
bytes.Repeat([]byte{0xAA}, 69), // present
nil, // absent
bytes.Repeat([]byte{0xBB}, 69), // present
},
}
payload := encodeToBytes(in.Encode)
var out AtomicGetResponse
if err := out.Decode(NewReader(payload)); err != nil {
t.Fatalf("decode: %v", err)
}
if len(out.Values) != 3 {
t.Fatalf("values len = %d, want 3 (positional alignment lost)", len(out.Values))
}
if len(out.Values[1]) != 0 {
t.Fatalf("absent value decoded as %d bytes, want 0", len(out.Values[1]))
}
if !bytes.Equal(out.Values[0], in.Values[0]) || !bytes.Equal(out.Values[2], in.Values[2]) {
t.Fatal("present values did not survive the round trip")
}
}
func TestAtomicApplyRequest_RoundTrip(t *testing.T) {
in := &AtomicApplyRequest{Requests: []*AtomicChainRequests{{
PeerChainID: bytes.Repeat([]byte{0xD1}, 32),
RemoveRequests: [][]byte{bytes.Repeat([]byte{0x11}, 32)},
PutRequests: []*AtomicElement{{
Key: bytes.Repeat([]byte{0x22}, 32),
Value: bytes.Repeat([]byte{0x33}, 69),
Traits: [][]byte{bytes.Repeat([]byte{0x44}, 20)},
}},
}}}
payload := encodeToBytes(in.Encode)
var out AtomicApplyRequest
if err := out.Decode(NewReader(payload)); err != nil {
t.Fatalf("decode: %v", err)
}
if len(out.Requests) != 1 {
t.Fatalf("requests len = %d, want 1", len(out.Requests))
}
got := out.Requests[0]
if !bytes.Equal(got.PeerChainID, in.Requests[0].PeerChainID) {
t.Fatal("peerChainID mismatch")
}
if len(got.RemoveRequests) != 1 || !bytes.Equal(got.RemoveRequests[0], in.Requests[0].RemoveRequests[0]) {
t.Fatal("removes mismatch")
}
if len(got.PutRequests) != 1 {
t.Fatalf("puts len = %d, want 1", len(got.PutRequests))
}
wantPut := in.Requests[0].PutRequests[0]
gotPut := got.PutRequests[0]
if !bytes.Equal(gotPut.Key, wantPut.Key) || !bytes.Equal(gotPut.Value, wantPut.Value) {
t.Fatal("put key/value mismatch")
}
if len(gotPut.Traits) != 1 || !bytes.Equal(gotPut.Traits[0], wantPut.Traits[0]) {
t.Fatal("put traits mismatch")
}
}
// TestAtomicApplyRequest_EmptyIsWellFormed: a block with no cross-chain ops must
// still encode/decode cleanly (the flush path skips the call in that case, but a
// zero-length request must never be a decode error).
func TestAtomicApplyRequest_EmptyIsWellFormed(t *testing.T) {
payload := encodeToBytes((&AtomicApplyRequest{}).Encode)
var out AtomicApplyRequest
if err := out.Decode(NewReader(payload)); err != nil {
t.Fatalf("decode of empty apply: %v", err)
}
if len(out.Requests) != 0 {
t.Fatalf("requests len = %d, want 0", len(out.Requests))
}
}
// --- InitializeRequest skew. These are the rollout-risk tests: a wire-format
// change that forks a mixed fleet is the failure mode this whole design is built
// to avoid, so both skew directions are pinned here rather than reasoned about.
// TestInitializeRequest_OldDecoderIgnoresNewTrailingFields is the NEW node ->
// OLD plugin direction. The old decoder stops after ServerAddr; the appended
// bytes are inert. Simulated by decoding only through ServerAddr and asserting
// no error and that the earlier fields are intact.
func TestInitializeRequest_OldDecoderIgnoresNewTrailingFields(t *testing.T) {
newReq := &InitializeRequest{
NetworkID: 3,
ChainID: bytes.Repeat([]byte{0xC1}, 32),
NodeID: bytes.Repeat([]byte{0x0E}, 20),
PublicKey: []byte{0x01},
XChainID: bytes.Repeat([]byte{0x02}, 32),
CChainID: bytes.Repeat([]byte{0x03}, 32),
UTXOAssetID: bytes.Repeat([]byte{0x04}, 32),
ChainDataDir: "/data/chain",
GenesisBytes: []byte("genesis"),
UpgradeBytes: []byte("upgrade"),
ConfigBytes: []byte("config"),
DBServerAddr: "127.0.0.1:1",
ServerAddr: "127.0.0.1:2",
AtomicServerAddr: "127.0.0.1:3",
DChainID: bytes.Repeat([]byte{0xD1}, 32),
}
payload := encodeToBytes(newReq.Encode)
// Replay the pre-change field order exactly — this is what an old plugin does.
r := NewReader(payload)
readBytes := func() []byte { v, err := r.ReadBytes(); mustNoErr(t, err); return v }
readStr := func() string { v, err := r.ReadString(); mustNoErr(t, err); return v }
netID, err := r.ReadUint32()
mustNoErr(t, err)
if netID != 3 {
t.Fatalf("networkID = %d, want 3", netID)
}
chainID := readBytes()
readBytes() // nodeID
readBytes() // publicKey
readBytes() // xChainID
readBytes() // cChainID
readBytes() // utxoAssetID
dataDir := readStr()
readBytes() // genesis
readBytes() // upgrade
readBytes() // config
readStr() // dbServerAddr
serverAddr := readStr()
if !bytes.Equal(chainID, newReq.ChainID) {
t.Fatal("old decoder mis-read ChainID")
}
if dataDir != "/data/chain" || serverAddr != "127.0.0.1:2" {
t.Fatalf("old decoder mis-read tail: dataDir=%q serverAddr=%q", dataDir, serverAddr)
}
// The appended fields are still on the wire — an old decoder simply never
// looks at them. That inertness is the point.
if r.Remaining() == 0 {
t.Fatal("expected trailing appended bytes to be present but unread")
}
}
// TestInitializeRequest_NewDecoderReadsOldPayload is the OLD node -> NEW plugin
// direction: the payload ends after ServerAddr, and the new decoder must yield
// an ABSENT atomic capability rather than io.ErrUnexpectedEOF. This is the exact
// skew class that broke every EVM Initialize with "zap decode initialize
// response: unexpected EOF" when Capabilities was appended.
func TestInitializeRequest_NewDecoderReadsOldPayload(t *testing.T) {
// Build a payload in the OLD field order (no appended fields).
old := encodeToBytes(func(buf *Buffer) {
buf.WriteUint32(3)
buf.WriteBytes(bytes.Repeat([]byte{0xC1}, 32))
buf.WriteBytes(bytes.Repeat([]byte{0x0E}, 20))
buf.WriteBytes([]byte{0x01})
buf.WriteBytes(bytes.Repeat([]byte{0x02}, 32))
buf.WriteBytes(bytes.Repeat([]byte{0x03}, 32))
buf.WriteBytes(bytes.Repeat([]byte{0x04}, 32))
buf.WriteString("/data/chain")
buf.WriteBytes([]byte("genesis"))
buf.WriteBytes([]byte("upgrade"))
buf.WriteBytes([]byte("config"))
buf.WriteString("127.0.0.1:1")
buf.WriteString("127.0.0.1:2")
})
var out InitializeRequest
if err := out.Decode(NewReader(old)); err != nil {
t.Fatalf("new decoder on old payload: %v (this is the EOF regression)", err)
}
if out.AtomicServerAddr != "" {
t.Fatalf("AtomicServerAddr = %q, want empty (capability must read as ABSENT)", out.AtomicServerAddr)
}
if len(out.DChainID) != 0 {
t.Fatalf("DChainID = %x, want empty", out.DChainID)
}
if out.ServerAddr != "127.0.0.1:2" || out.ChainDataDir != "/data/chain" {
t.Fatal("new decoder mis-read the old payload's existing fields")
}
}
func TestInitializeRequest_RoundTripWithAtomicFields(t *testing.T) {
in := &InitializeRequest{
NetworkID: 1,
ChainID: bytes.Repeat([]byte{0xC1}, 32),
NodeID: bytes.Repeat([]byte{0x0E}, 20),
XChainID: bytes.Repeat([]byte{0x02}, 32),
CChainID: bytes.Repeat([]byte{0x03}, 32),
UTXOAssetID: bytes.Repeat([]byte{0x04}, 32),
ChainDataDir: "/d",
AtomicServerAddr: "/tmp/atomic.sock",
DChainID: bytes.Repeat([]byte{0xD1}, 32),
}
var out InitializeRequest
if err := out.Decode(NewReader(encodeToBytes(in.Encode))); err != nil {
t.Fatalf("decode: %v", err)
}
if out.AtomicServerAddr != in.AtomicServerAddr {
t.Fatalf("AtomicServerAddr = %q, want %q", out.AtomicServerAddr, in.AtomicServerAddr)
}
if !bytes.Equal(out.DChainID, in.DChainID) {
t.Fatalf("DChainID = %x, want %x", out.DChainID, in.DChainID)
}
}
func mustNoErr(t *testing.T, err error) {
t.Helper()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
+61 -3
View File
@@ -46,6 +46,29 @@ type InitializeRequest struct {
ConfigBytes []byte
DBServerAddr string
ServerAddr string
// AtomicServerAddr is the address of a ZAP server the node bound over THIS
// chain's atomic shared-memory handle before calling Initialize. A plugin
// dials it to reach the primary network's cross-chain import/export
// primitive, which cannot be copied into the plugin's rebuilt Runtime
// because it is an interface over a live database the node owns. Same shape
// and lifetime as DBServerAddr.
//
// Empty means the node wired no shared memory for this chain. The plugin
// MUST then leave Runtime.SharedMemory nil so a settlement precompile
// reverts fail-closed rather than fabricate value.
AtomicServerAddr string
// DChainID is the D-Chain (dexvm) blockchain id, resolved node-side from the
// chain alias "D". It rides here because the plugin's Runtime has no
// BCLookup — that field is an interface onto the node's chain manager and,
// like SharedMemory, does not survive the boundary. The DEX settlement seam
// needs exactly one alias resolved, so the resolved ID is carried instead of
// proxying a whole lookup service.
//
// Empty means the network has no dexvm deployed; the settlement seam then
// stays closed rather than guess a peer.
DChainID []byte
}
// Encode serializes InitializeRequest to the buffer
@@ -63,9 +86,29 @@ func (m *InitializeRequest) Encode(buf *Buffer) {
buf.WriteBytes(m.ConfigBytes)
buf.WriteString(m.DBServerAddr)
buf.WriteString(m.ServerAddr)
// OPTIONAL trailing fields — see the EVOLUTION RULE on InitializeResponse.
// Appended, never inserted, so an older decoder stops after ServerAddr and
// ignores these bytes.
buf.WriteString(m.AtomicServerAddr)
buf.WriteBytes(m.DChainID)
}
// Decode deserializes InitializeRequest from the reader
// Decode deserializes InitializeRequest from the reader.
//
// LENGTH-TOLERANT trailing fields, the same contract InitializeResponse.Decode
// documents. AtomicServerAddr and DChainID were APPENDED for the cross-chain
// atomic seam without bumping version.RPCChainVMProtocol, so both skew
// directions stay safe:
//
// - NEW node -> OLD plugin: the old decoder stops after ServerAddr and ignores
// the trailing bytes (ZAP frames are length-prefixed). The plugin keeps its
// nil SharedMemory and its settlements keep reverting fail-closed — the
// behavior it already had, never a silent divergence.
// - OLD node -> NEW plugin: Remaining() is 0 at these reads, so both fields
// stay zero-valued and the plugin treats the atomic capability as absent.
//
// Neither direction changes how any block executes, which is what makes this
// safe to roll one validator at a time.
func (m *InitializeRequest) Decode(r *Reader) error {
var err error
if m.NetworkID, err = r.ReadUint32(); err != nil {
@@ -104,8 +147,23 @@ func (m *InitializeRequest) Decode(r *Reader) error {
if m.DBServerAddr, err = r.ReadString(); err != nil {
return err
}
m.ServerAddr, err = r.ReadString()
return err
if m.ServerAddr, err = r.ReadString(); err != nil {
return err
}
// OPTIONAL trailing fields. A length-prefixed field costs at least its
// 4-byte count, so Remaining() < 4 ⇒ the peer predates this field; leave it
// zero-valued instead of returning io.ErrUnexpectedEOF.
if r.Remaining() >= 4 {
if m.AtomicServerAddr, err = r.ReadString(); err != nil {
return err
}
}
if r.Remaining() >= 4 {
if m.DChainID, err = r.ReadBytes(); err != nil {
return err
}
}
return nil
}
// InitializeResponse contains initialization results
+18
View File
@@ -64,6 +64,24 @@ const (
MsgBlockReject
MsgStateSummaryAccept // = 31
// Atomic shared-memory methods (32-39). The cross-chain atomic import/
// export primitive (the SAME one platformvm and the dexvm use) lives in the
// NODE process — one atomic.Memory per node, shared by every chain. A VM
// hosted OUT of process (the C-Chain EVM, the dexvm) therefore cannot reach
// it through its Runtime, which is why these exist.
//
// DIRECTION: unlike every other message in this file, these travel
// plugin->node. The plugin DIALS a per-chain ZAP server the node binds
// before Initialize and names in InitializeRequest.AtomicServerAddr — the
// SAME shape as DBServerAddr. The node is the server; the transport stays
// strictly request/response and needs no duplex.
//
// SCOPE: Get and Apply only. SharedMemory.Indexed is deliberately NOT
// proxied — no settlement path calls it, and an unused wire surface is a
// liability, not a feature.
MsgAtomicGet MessageType = 32
MsgAtomicApply MessageType = 33
// p2p.Sender methods (40-49)
MsgSendRequest MessageType = 40
MsgSendResponse MessageType = 41