Documentation
¶
Overview ¶
Package dataplane is the worker↔coord transport for high-rate task traffic that NATS JetStream is a poor fit for at scale: task dispatch, result batches, gather payloads, and per-task progress. Out of scope (stays on NATS): heartbeats, query cancellation, catalog/UDF KV, DLQ.
See project_split_plane_design_2026-05-20.md for the design.
Phase A: connection bootstrap only (Hello/Welcome). The streaming RPCs that carry tasks and results are added in Phases B–E.
Index ¶
- Variables
- type Client
- func (c *Client) ClusterID() string
- func (c *Client) Connected() bool
- func (c *Client) RegisterDispatchHandler(h DispatchHandler)
- func (c *Client) SendResultBatch(rb ResultBatch) error
- func (c *Client) SendTaskProgress(tp TaskProgress) error
- func (c *Client) Start(ctx context.Context)
- func (c *Client) Stop()
- type ClientConfig
- type DispatchHandler
- type PeerClient
- type PeerServer
- type PeerServerConfig
- type ResultBatch
- type ResultHandler
- type Server
- func (s *Server) Addr() string
- func (s *Server) Connect(stream grpc.BidiStreamingServer[dpv1.WorkerEnvelope, dpv1.CoordEnvelope]) error
- func (s *Server) ConnectedWorkers() []string
- func (s *Server) PickWorker() (string, bool)
- func (s *Server) RegisterResultHandler(queryID string, h ResultHandler)
- func (s *Server) RegisterTaskProgressHandler(queryID string, h TaskProgressHandler)
- func (s *Server) SendTaskDispatch(workerID, taskID, queryID, stageID string, taskBlob []byte, ...) error
- func (s *Server) SetGlobalTaskProgressHandler(h TaskProgressHandler)
- func (s *Server) Start() error
- func (s *Server) Stop(gracePeriod time.Duration)
- func (s *Server) UnregisterResultHandler(queryID string)
- func (s *Server) UnregisterTaskProgressHandler(queryID string)
- type ServerConfig
- type ShuffleFileResolver
- type TaskDispatch
- type TaskProgress
- type TaskProgressHandler
Constants ¶
This section is empty.
Variables ¶
var ( ErrPeerDenied = errors.New("peer fetch denied") ErrPeerNotFound = errors.New("peer file not found") )
Sentinel errors for ShuffleFileResolver implementations.
var ErrNoWorkers = errors.New("dataplane: no connected workers")
ErrNoWorkers is returned by SendTaskDispatch when no worker is currently connected to the data-plane server.
var ErrNotConnected = errors.New("dataplane: not connected")
ErrNotConnected is returned by SendResultBatch (and future send methods) when the data-plane stream is not currently usable. Callers can choose to fall back to NATS or treat it as fatal.
var PeerFetchIdleTimeout = 15 * time.Second
PeerFetchIdleTimeout bounds the wait for each stream chunk. The server streams from local NVMe, so inter-chunk gaps are normally microseconds; a stream that stops delivering WITHOUT erroring is a wedged or half-dead peer, and it must become an error the tiered read path can fall through on — the 2026-08-11 Q21-R2 stall was one such silent stream holding an eager hash-join build (and with it the whole cluster's barrier) for 228s, invisible to every phase timer and to the tier fallback, which only reacts to errors. Var so tests can shrink it.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is the worker-side data-plane gRPC connection. It maintains a single long-lived bidi stream to coord, reconnecting on disconnect.
Phase A: open the stream, exchange Hello/Welcome, hold open. Phase B: send ResultBatch messages via the stream's send side. Phase C: route TaskDispatch from the recv side into a registered handler (worker plumbs this into its task pool).
func NewClient ¶
func NewClient(cfg ClientConfig, logger *slog.Logger) *Client
NewClient constructs a Client. Call Start to dial.
func (*Client) ClusterID ¶
ClusterID returns the coord-reported cluster identity, or "" if not yet connected.
func (*Client) Connected ¶
Connected reports whether the most recent connect attempt produced a live stream that has received Welcome.
func (*Client) RegisterDispatchHandler ¶
func (c *Client) RegisterDispatchHandler(h DispatchHandler)
RegisterDispatchHandler installs the function invoked for each TaskDispatch envelope arriving on the stream. Must be set before coord starts pushing work. Replacing the handler at runtime is safe — the recv loop reads atomically per message. A nil handler is rejected (atomic.Value disallows storing nil).
func (*Client) SendResultBatch ¶
func (c *Client) SendResultBatch(rb ResultBatch) error
SendResultBatch sends a result chunk over the data-plane stream. Returns ErrNotConnected if the stream is not currently usable. Callers can choose to fall back to NATS or surface the error. Safe for concurrent callers — gRPC stream.Send is serialized internally via sendMu.
func (*Client) SendTaskProgress ¶
func (c *Client) SendTaskProgress(tp TaskProgress) error
SendTaskProgress emits a TaskProgress envelope on the data-plane stream. Cheap fire-and-forget: a missed progress message doesn't break correctness (worker-side ticker re-sends in ~2 s). Returns ErrNotConnected when the stream isn't live so callers can decide whether to log or drop.
type ClientConfig ¶
type ClientConfig struct {
// CoordAddr is the coord's data-plane endpoint (host:port). Required.
CoordAddr string
// WorkerID is sent to coord in Hello. Required.
WorkerID string
// BuildSHA is sent to coord for version-skew detection.
BuildSHA string
// TLSConfig enables TLS when non-nil. Must match the server's choice.
TLSConfig *tls.Config
// ReconnectBackoff is the initial backoff after disconnect.
// Default: 500ms, doubling up to 10s.
ReconnectBackoff time.Duration
}
ClientConfig configures the worker-side gRPC dialer.
type DispatchHandler ¶
type DispatchHandler func(TaskDispatch)
DispatchHandler consumes TaskDispatch messages. The recv loop calls the handler synchronously; a slow handler applies backpressure all the way back to coord via HTTP/2 flow control.
type PeerClient ¶
type PeerClient struct {
// contains filtered or unexported fields
}
PeerClient dials worker PeerExchange endpoints and streams shuffle files. One cached ClientConn per peer address; connections are lazy (grpc.NewClient) and RPCs fail fast when a peer is unreachable — the caller falls through to S3, so a dead peer costs one failed dial, not a stall.
func NewPeerClient ¶
func NewPeerClient(tlsConfig *tls.Config) *PeerClient
NewPeerClient constructs a PeerClient. tlsConfig nil = plaintext, matching the peer server's default intra-cluster posture.
func (*PeerClient) FetchShuffle ¶
func (c *PeerClient) FetchShuffle(ctx context.Context, addr, queryID, key, token string) (io.ReadCloser, error)
FetchShuffle opens a fetch stream for (queryID, key) against the peer at addr and returns a reader over the file bytes. The returned reader must be closed; closing cancels the stream. Errors — dial failure, PermissionDenied, NotFound, mid-stream disconnect, or an inter-chunk gap exceeding PeerFetchIdleTimeout — surface from Read; the caller treats any of them as a cache miss and falls through to S3.
type PeerServer ¶
type PeerServer struct {
dpv1.UnimplementedPeerExchangeServer
// contains filtered or unexported fields
}
PeerServer is the worker-side PeerExchange gRPC listener. Serving is a resolver lookup + chunked file copy; all state lives in the resolver.
func NewPeerServer ¶
func NewPeerServer(cfg PeerServerConfig, resolver ShuffleFileResolver, logger *slog.Logger) *PeerServer
NewPeerServer constructs a PeerServer. Call Start to begin accepting.
func (*PeerServer) AdvertiseAddr ¶
func (s *PeerServer) AdvertiseAddr() string
AdvertiseAddr returns the address peers should dial, for the worker to carry in its heartbeats. Empty until Start has bound the listener (unless an explicit AdvertiseAddr was configured).
The derived address is resolved ONCE and cached. Deriving it means a netlink RIB dump (net.InterfaceAddrs → NetlinkRIB), and netlink route dumps serialize on the kernel's global rtnl_lock — held by ec2net policy- route refreshes and ENA reconfiguration on EC2. The heartbeat loop calls this every tick; before the cache, a wedged rtnl_lock stalled the heartbeat goroutine inside the kernel, silencing heartbeats AND peer advertisement while the process stayed alive — the dispatch-stall arc's "network-silent while alive" family (2026-08-13 frozen-spin stack capture caught the heartbeat goroutine in syscall.NetlinkRIB/bind). The address cannot change after the listener binds, so the dump buys nothing.
func (*PeerServer) FetchShuffle ¶
func (s *PeerServer) FetchShuffle(req *dpv1.FetchShuffleRequest, stream grpc.ServerStreamingServer[dpv1.ShuffleChunk]) error
FetchShuffle implements PeerExchangeServer: resolve the key to a local file and stream it in peerChunkBytes frames. Backpressure per stream is HTTP/2 flow control; the semaphore only bounds fan-in concurrency.
func (*PeerServer) Start ¶
func (s *PeerServer) Start() error
Start binds the listener and runs the gRPC server in the background.
func (*PeerServer) Stop ¶
func (s *PeerServer) Stop()
Stop shuts the server down immediately. In-flight fetches abort; their consumers fall through to S3.
type PeerServerConfig ¶
type PeerServerConfig struct {
// Addr is the listen address (e.g. ":9095", or ":0" for tests). Required.
Addr string
// AdvertiseAddr is the externally-dialable address peers use, carried in
// worker heartbeats. When empty it is derived from the bound listener:
// the listener's host if it is a specific IP, else the first
// non-loopback unicast IPv4 (falling back to 127.0.0.1).
AdvertiseAddr string
// TLSConfig enables TLS when non-nil. Default (nil) is plaintext —
// matching the coord data plane's intra-cluster trust posture.
TLSConfig *tls.Config
// MaxConcurrentFetches caps concurrently-served FetchShuffle streams,
// protecting the producer's NVMe/NIC from consumer fan-in spikes.
// 0 = default (16).
MaxConcurrentFetches int
// CompressWire s2-compresses raw WSHF payloads on the stream
// (docs/design/peer-wire-compression.md): the served bytes become a
// standard WSHC envelope, which every consumer already decodes.
// Payloads that are already WSHC (or not WSHF at all) pass through
// untouched. Trades ~1 core-GB/s of producer CPU per stream for
// ~20% fewer wire bytes (the s2 ratio measured on SF100 shuffle
// data). Default false pending validation.
CompressWire bool
}
PeerServerConfig configures a worker's PeerExchange listener.
type ResultBatch ¶
type ResultBatch struct {
QueryID string
WorkerID string
Terminal bool
RowCount int32
Payload []byte // WSHF-encoded
Err string
}
ResultBatch is the transport-neutral form of a worker→coord result chunk. The gRPC adapter translates ResultBatch proto messages into this struct before invoking a registered ResultHandler; coord owns the handler and routes by QueryID. Mirrors distributed.GatherBatchMsg minus the QueryID (which gRPC needs for routing — NATS gets it from the subject).
type ResultHandler ¶
type ResultHandler func(*ResultBatch)
ResultHandler consumes ResultBatch messages for a single query. Registered by coord with Server.RegisterResultHandler at query start and removed via UnregisterResultHandler at query end. Must be thread-safe — multiple workers may stream concurrently to the same query.
type Server ¶
type Server struct {
dpv1.UnimplementedDataPlaneServer
// contains filtered or unexported fields
}
Server is the coord-side data-plane gRPC listener. Workers dial it and open one long-lived bidi stream each.
func NewServer ¶
func NewServer(cfg ServerConfig, logger *slog.Logger) *Server
NewServer constructs a Server. Call Start to begin accepting.
func (*Server) Connect ¶
func (s *Server) Connect(stream grpc.BidiStreamingServer[dpv1.WorkerEnvelope, dpv1.CoordEnvelope]) error
Connect implements DataPlaneServer. Phase A only handshakes Hello → Welcome and holds the stream open until the worker closes. Phases B–E route TaskDispatch outbound and consume the worker's result/progress messages.
func (*Server) ConnectedWorkers ¶
ConnectedWorkers returns a snapshot of worker IDs currently registered. Order is registration order — useful for tests; the scheduler uses PickWorker for round-robin dispatch.
func (*Server) PickWorker ¶
PickWorker returns the next worker ID via round-robin across the currently-connected set. Returns ("", false) when no worker is connected. Safe for concurrent callers.
func (*Server) RegisterResultHandler ¶
func (s *Server) RegisterResultHandler(queryID string, h ResultHandler)
RegisterResultHandler installs a per-query handler. Multiple concurrent registrations for the same queryID overwrite the previous one (workers can only stream to whichever handler is current). Must be paired with UnregisterResultHandler when the query finishes.
func (*Server) RegisterTaskProgressHandler ¶
func (s *Server) RegisterTaskProgressHandler(queryID string, h TaskProgressHandler)
RegisterTaskProgressHandler installs the per-query progress handler. Each query overrides any prior registration for the same queryID. Must be paired with UnregisterTaskProgressHandler at query end.
func (*Server) SendTaskDispatch ¶
func (s *Server) SendTaskDispatch(workerID, taskID, queryID, stageID string, taskBlob []byte, deadlineUnixNano int64) error
SendTaskDispatch pushes a TaskDispatch envelope to the named worker. Blocks on HTTP/2 flow control if the worker's recv side is full — that's the backpressure path. Returns ErrNotConnected if the worker is no longer connected.
task_blob is the result of distributed.Marshal(Task). The scheduler owns the marshal; the data-plane layer only carries bytes.
func (*Server) SetGlobalTaskProgressHandler ¶
func (s *Server) SetGlobalTaskProgressHandler(h TaskProgressHandler)
SetGlobalTaskProgressHandler installs (or clears, when h is nil) the global TaskProgress handler. Used by WorkerRegistry to drive the multi-signal liveness path off gRPC TaskProgress arrivals.
func (*Server) Start ¶
Start binds the listener and runs the gRPC server in the background. Returns once the listener is bound; the serve loop runs until Stop.
func (*Server) Stop ¶
Stop shuts the server down. Tries graceful first (waits up to gracePeriod for in-flight RPCs to finish), then forces if any are still active. A gracePeriod of 0 means force immediately, useful for tests where worker streams stay open by design.
func (*Server) UnregisterResultHandler ¶
UnregisterResultHandler removes the handler. Late-arriving batches after this call are dropped. Safe to call when no handler was ever registered.
func (*Server) UnregisterTaskProgressHandler ¶
UnregisterTaskProgressHandler removes the per-query handler.
type ServerConfig ¶
type ServerConfig struct {
// Addr is the listen address (e.g. ":9091"). Required.
Addr string
// ClusterID identifies this cluster; sent to workers in Welcome.
ClusterID string
// TLSConfig enables TLS when non-nil. Default (nil) is plaintext —
// intra-cluster traffic on a trusted private network. Set for
// less-trusted deployments. NATS control plane keeps its own mTLS
// regardless of this setting.
TLSConfig *tls.Config
// MaxConcurrentStreams caps connected workers. 0 = unlimited.
MaxConcurrentStreams uint32
}
ServerConfig configures the coord-side gRPC listener.
type ShuffleFileResolver ¶
type ShuffleFileResolver interface {
ResolveShuffleFile(ctx context.Context, queryID, key, token string) (string, error)
}
ShuffleFileResolver maps a fetch request to a local file path. The worker implements it over its LocalStageCache + per-query fetch tokens. Implementations return ErrPeerDenied for a bad token and ErrPeerNotFound for a key the worker doesn't hold; both are terminal for the fetch and the consumer falls through to S3. ctx is the fetch stream's context — a resolver that does real work (base-table owner read-through) bounds its wait on it.
type TaskDispatch ¶
type TaskDispatch struct {
TaskID string
QueryID string
StageID string
TaskBlob []byte
DeadlineUnixNano int64
}
TaskDispatch is the transport-neutral form of a coord→worker task envelope. The worker registers a handler with RegisterDispatchHandler; every TaskDispatch received from the stream is delivered to it.
Body is the same `distributed.Marshal(Task)` blob coord publishes — worker unmarshals via `distributed.Unmarshal` to recover the full Task.
type TaskProgress ¶
type TaskProgress struct {
QueryID string
StageID string
TaskID string
WorkerID string
RowsProcessed int64
BytesProcessed int64
TimestampUnixNano int64
}
TaskProgress is the transport-neutral form of a worker→coord progress update. Mirrors distributed.TaskProgress (the NATS path's shape) minus the JSON tags so coord-side bridges can reuse the existing handler bodies.
type TaskProgressHandler ¶
type TaskProgressHandler func(*TaskProgress)
TaskProgressHandler consumes TaskProgress messages. Server invokes the global handler (used by WorkerRegistry for liveness) AND the per-query handler (used by stageProgressBridge) on every arrival; both are optional.
Source Files