Direct Messaging

The messaging protocol provides request-response semantics for targeted peer-to-peer communication. Unlike GossipSub which broadcasts to all peers, messaging sends a request to a specific peer and waits for a response.

Protocol ID: /xe/msg/1.0.0

Wire Types

MsgRequest

type MsgRequest struct {
    Type    string          `json:"type"`
    Payload json.RawMessage `json:"payload"`
}

MsgResponse

type MsgResponse struct {
    Type    string          `json:"type"`
    Error   string          `json:"error,omitempty"`
    Payload json.RawMessage `json:"payload"`
}

MsgHandler

type MsgHandler func(from peer.ID, payload json.RawMessage) (json.RawMessage, error)

Handlers receive the sender's peer ID and the raw JSON payload. They return a raw JSON response or an error.

Constants

Messenger

The Messenger struct manages handler registration and request dispatching:

type Messenger struct {
    host     host.Host
    dht      *dht.IpfsDHT
    handlers map[string]MsgHandler
    mu       sync.RWMutex
}

Construction

func NewMessenger(h host.Host, d *dht.IpfsDHT) *Messenger

NewMessenger creates a Messenger and registers the /xe/msg/1.0.0 stream handler on the host. The DHT is used for peer discovery when the target peer is not in the peerstore.

Registering Handlers

func (m *Messenger) Handle(msgType string, h MsgHandler)

Registers a handler for a specific message type. Handlers are stored in a thread-safe map and dispatched by the Type field of incoming requests.

[!NOTE] Note If a request arrives with an unregistered message type, the Messenger responds with an error: "unknown message type: {type}".

Sending Requests

func (m *Messenger) Request(ctx context.Context, target peer.ID, msgType string, payload any) (json.RawMessage, error)

Sends a typed request to a specific peer and waits for the response. The flow:

1. Marshal payload -- The payload (any type) is JSON-marshaled into json.RawMessage
2. Find peer -- If the target is not in the peerstore and a DHT is available, FindPeer() is called to locate the peer and Connect() establishes a connection
3. Open stream -- A new stream is opened to the target on protocol /xe/msg/1.0.0
4. Set deadline -- Uses the context deadline if set, otherwise defaults to 30 seconds
5. Send request -- JSON-encodes the MsgRequest and calls CloseWrite() to signal completion
6. Read response -- JSON-decodes the MsgResponse from a size-limited reader
7. Check error -- If the response contains an Error field, returns it as a Go error

Request-Response Flow

Client                                    Server
──────                                    ──────
  │                                          │
  │  Request(ctx, target, type, payload)     │
  │                                          │
  │  1. Find peer via DHT (if needed)        │
  │  2. Connect (if needed)                  │
  │  3. Open stream                          │
  │                                          │
  │  MsgRequest{Type, Payload} ────────────▶ │
  │  CloseWrite() ─────────────────────────▶ │
  │                                          │  Lookup handler by Type
  │                                          │  Call handler(from, payload)
  │                                          │
  │  ◀──────────── MsgResponse{Type, Payload}│
  │                                          │
  │  Close stream                            │
  │                                          │

Peer Discovery via DHT

When Request() is called for a peer not in the peerstore, the Messenger uses the Kademlia DHT to find the peer:

if len(m.host.Peerstore().Addrs(target)) == 0 && m.dht != nil {
    pi, err := m.dht.FindPeer(ctx, target)
    // ... connect to discovered peer
}

This makes messaging work even when the sender has never directly connected to the recipient, as long as both are part of the DHT.

Message Types

The messaging protocol is generic -- the Type field determines which handler processes the request. The following message types are registered as point-to-point handlers:

[!NOTE] Gossip vs messaging Block propagation, votes, marketplace negotiations, directory updates, certificates, and state chain blocks are all broadcast over GossipSub (pubsub topics), not the point-to-point messaging protocol. See Gossip. State chain catch-up uses its own stream protocol, /xe/statechain-sync/1.0.0, not the Messenger.

Targeted Pulls

Three of the message types above exist because broadcast alone is not always enough to un-stick a node. All three are best-effort and idempotent: a peer that does not answer or holds nothing is skipped, and the caller retries on its next sweep.

block_request — pull a block body by hash

type BlockRequest struct {
    Hashes []string `json:"hashes"`
}

type BlockResponse struct {
    Blocks []*core.Block `json:"blocks"`
}

A node can end up naming a block in a conflict record whose body it never received: the block was not re-gossiped, and the sync server's anti-amplification guard refuses to backfill a frontier it cannot recognise. Without both sibling bodies the node cannot run weighted two-candidate voting, so the account stalls. PullBlocks asks peers for the missing bodies by hash and stops as soon as every requested hash has been obtained.

vote_request — pull the votes at a position

type VoteRequest struct {
    Account  string `json:"account"`
    Previous string `json:"previous"`
}

type VoteResponse struct {
    Votes []*core.Vote `json:"votes"`
}

Frontier sync carries block bodies but no finalization state. A node holding its own non-final losing fork stages the network's winner as a conflict it can never resolve — the network already finalized that winner, and the post-finalization cleanup deleted every representative's stored votes, so those votes are never re-gossiped. PullVotes fetches them directly; the answering node adds a freshly-derived final vote when it has already finalized the child at that position. Ingested votes re-drive the local tally, and the winner is then promoted through the normal reorg path — whose finality wall is the safety backstop.

cert_request — pull provider certificates

A node that restarts or joins late needs provider performance certificates to validate certificate-referencing lease blocks. Rather than waiting for the next gossip rebroadcast, it pulls them from each peer on connect (retrying up to three times with backoff), plus a background sweep of one rotating peer every 2 minutes. A response is capped at 256 certificates.

Payload Types

The marketplace payload types below are defined alongside the messaging types in net/messages.go, but they travel over the marketplace gossip topic, not the Messenger.

ResourceAdvertisement

type ResourceAdvertisement struct {
    Provider            string `json:"provider"`
    VCPUs               uint64 `json:"vcpus"`
    MemoryMB            uint64 `json:"memory_mb"`
    DiskGB              uint64 `json:"disk_gb"`
    MaxConcurrentLeases uint64 `json:"max_concurrent_leases,omitempty"`
    Timestamp           int64  `json:"timestamp"`
    Signature           string `json:"signature"`
}

ResourceRequest

type ResourceRequest struct {
    Consumer  string `json:"consumer"`
    RequestID string `json:"request_id"`
    VCPUs     uint64 `json:"vcpus"`
    MemoryMB  uint64 `json:"memory_mb"`
    DiskGB    uint64 `json:"disk_gb"`
    Duration  uint64 `json:"duration"`  // seconds
    Timestamp int64  `json:"timestamp"`
    Signature string `json:"signature"`
}

ResourceOffer

type ResourceOffer struct {
    Provider        string `json:"provider"`
    RequestID       string `json:"request_id"`
    VCPUs           uint64 `json:"vcpus"`
    MemoryMB        uint64 `json:"memory_mb"`
    DiskGB          uint64 `json:"disk_gb"`
    Duration        uint64 `json:"duration"`
    TotalCost       uint64 `json:"total_cost"`
    CertificateHash string `json:"certificate_hash,omitempty"`
    Timestamp       int64  `json:"timestamp"`
    Signature       string `json:"signature"`
}

CertificateHash pins the provider's performance certificate so the consumer can lock in the price multiplier at lease creation.

MarketplaceMsg

type MarketplaceMsg struct {
    Type    string                 `json:"type"` // "advertisement", "request", "offer"
    Ad      *ResourceAdvertisement `json:"ad,omitempty"`
    Request *ResourceRequest       `json:"request,omitempty"`
    Offer   *ResourceOffer         `json:"offer,omitempty"`
}

Error Handling

Errors can occur at multiple levels:

Comparison with GossipSub

Use messaging when you need a response from a specific peer. Use gossip when you need to broadcast to the entire network.