Documentation
¶
Overview ¶
Package harmonypeerhttp implements the HTTP transport for harmonytask's peer-to-peer messaging protocol. Each Curio node runs this as an HTTP handler mounted at /peer/v1. When a peer sends a task event (new task, reservation, or start notification), it arrives as an HTTP POST with the sender's address in the X-Peer-ID header.
Design choices:
- Stateless HTTP POST (not WebSocket) keeps the transport simple and compatible with standard HTTP infrastructure (load balancers, proxies).
- Each message is a single POST; there is no long-lived connection. The "connection" abstraction is maintained in-memory: the first POST from a new peer triggers OnConnect, creating a peerHTTPConnection that feeds subsequent POSTs into a buffered channel.
- Failed sends immediately drop the peer (no retries). This is intentional: the DB poll fallback ensures correctness, and retrying would add latency to the scheduler's fire-and-forget send path.
- The 2-second send timeout prevents a slow peer from blocking the sender.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type PeerHTTP ¶
type PeerHTTP struct {
// contains filtered or unexported fields
}
PeerHTTP is both an HTTP handler (for receiving peer messages) and a PeerConnectorInterface implementation (for sending messages to peers). It maintains an in-memory registry of peer connections, each backed by a buffered channel that decouples HTTP request handling from the peering layer's receive loop.
func New ¶
New creates a PeerHTTP instance. localAddr is this node's host:port, used to identify ourselves in the X-Peer-ID header on outbound messages.
func (*PeerHTTP) ConnectToPeer ¶
func (p *PeerHTTP) ConnectToPeer(peerID string) (harmonytask.PeerConnection, error)
ConnectToPeer creates or returns an existing connection to a peer. Unlike the test pipe implementation, HTTP connections are lazy: no actual network call happens here, so this method does not surface dial failures (harmonytask falls back to pollFrequently on handshake/send errors in handlePeer instead). The first SendMessage will make the HTTP POST, and the remote's ServeHTTP will trigger OnConnect.
func (*PeerHTTP) ServeHTTP ¶
func (p *PeerHTTP) ServeHTTP(w http.ResponseWriter, r *http.Request)
ServeHTTP handles incoming peer HTTP POST messages. Mount this at "/peer/v1" on the node's HTTP router.
Message flow:
- Extract X-Peer-ID header to identify the sender.
- Look up the existing peerHTTPConnection for this sender, or create one if this is the first message from this peer.
- Push the message body into the connection's buffered channel.
- Only when the connection is newly created, trigger OnConnect which starts the peering layer's handlePeer goroutine for it. Subsequent POSTs from the same peer reuse the connection and do NOT re-run the handshake — otherwise every message would spawn a new handlePeer goroutine (which sends an identity message back and runs DB queries), producing an unbounded identity ping-pong and goroutine/DB-query storm between peers.
The 100-slot buffer prevents a burst of messages from blocking HTTP responses. If the buffer is full, the message is dropped with 503 — the DB poll fallback ensures eventual consistency.
func (*PeerHTTP) SetOnConnect ¶
func (p *PeerHTTP) SetOnConnect(onConnect func(peerAddr string, conn harmonytask.PeerConnection))
SetOnConnect registers the callback that the harmonytask peering layer provides. It is called when the first message arrives from a new peer, establishing the bidirectional "connection" for that peer.
Source Files