Direct Messaging

Peer-to-peer 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

1type MsgResponse struct {2    Type    string          `json:"type"`3    Error   string          `json:"error,omitempty"`4    Payload json.RawMessage `json:"payload"`5}

#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

Constant	Value	Description
`MsgProtocol`	`/xe/msg/1.0.0`	Stream protocol identifier
`MsgStreamDeadline`	30 seconds	Read/write timeout per stream
`MsgMaxRequestSize`	64 KB (65,536 bytes)	Maximum incoming request size
`MsgMaxResponseSize`	64 KB (65,536 bytes)	Maximum incoming response size

#Messenger

The Messenger struct manages handler registration and request dispatching:

1type Messenger struct {2    host     host.Host3    dht      *dht.IpfsDHT4    handlers map[string]MsgHandler5    mu       sync.RWMutex6}

#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.

#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:

Marshal payload -- The payload (any type) is JSON-marshaled into json.RawMessage
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
Open stream -- A new stream is opened to the target on protocol /xe/msg/1.0.0
Set deadline -- Uses the context deadline if set, otherwise defaults to 30 seconds
Send request -- JSON-encodes the MsgRequest and calls CloseWrite() to signal completion
Read response -- JSON-decodes the MsgResponse from a size-limited reader
Check error -- If the response contains an Error field, returns it as a Go error

#Request-Response Flow

diagram

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:

Type	Direction	Purpose
`vm_credentials`	Provider → Consumer	Delivers VM SSH connection details after provisioning
`vm_status`	Consumer → Provider	Queries current VM status for a lease
`account_chat`	Any → Any	Sends a chat message between accounts
`attest_timestamp`	Provider → Timekeeper	Requests a signed timestamp attestation for a lease

#Payload Types

#ResourceAdvertisement

1type ResourceAdvertisement struct {2    Provider  string `json:"provider"`3    VCPUs     uint64 `json:"vcpus"`4    MemoryMB  uint64 `json:"memory_mb"`5    DiskGB    uint64 `json:"disk_gb"`6    Timestamp int64  `json:"timestamp"`7    Signature string `json:"signature"`8}

#ResourceRequest

1type ResourceRequest struct {2    Consumer  string `json:"consumer"`3    RequestID string `json:"request_id"`4    VCPUs     uint64 `json:"vcpus"`5    MemoryMB  uint64 `json:"memory_mb"`6    DiskGB    uint64 `json:"disk_gb"`7    Duration  uint64 `json:"duration"`  // seconds8    Timestamp int64  `json:"timestamp"`9    Signature string `json:"signature"`10}

#ResourceOffer

1type ResourceOffer struct {2    Provider  string `json:"provider"`3    RequestID string `json:"request_id"`4    VCPUs     uint64 `json:"vcpus"`5    MemoryMB  uint64 `json:"memory_mb"`6    DiskGB    uint64 `json:"disk_gb"`7    Duration  uint64 `json:"duration"`8    TotalCost uint64 `json:"total_cost"`9    Timestamp int64  `json:"timestamp"`10    Signature string `json:"signature"`11}

#StateChainSyncRequest / Response

1type StateChainSyncRequest struct {2    TipIndex int64 `json:"tip_index"` // -1 if empty chain3}4 5type StateChainSyncResponse struct {6    Blocks  []*statechain.Block `json:"blocks"`7    HasMore bool                `json:"has_more"`8}

#MarketplaceMsg

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

#Error Handling

Errors can occur at multiple levels:

Level	Handling
Peer not found (DHT)	`Request()` returns error
Connection failure	`Request()` returns error
Stream open failure	`Request()` returns error
Timeout (30s deadline)	Stream read/write fails
Unknown message type	Server responds with error in `MsgResponse.Error`
Handler returns error	Server responds with error in `MsgResponse.Error`
Decode failure	Server logs error, no response sent

#Comparison with GossipSub

	Messaging	GossipSub
Target	Specific peer	All peers
Pattern	Request-response	Publish-subscribe
Delivery	Reliable (or error)	Best-effort
Use case	Marketplace negotiation, targeted sync	Block/vote broadcast
Size limit	64 KB	256 KB
Timeout	30 seconds	None (async)

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