Struct Sink 

pub struct Sink {
    pub port: usize,
    pub queue_size: usize,
    pub message_size: usize,
    pub backend_address: String,
    pub metadata: WorkerMetadataSender,
}

Specifies the binding and operation parameters for a ZMQ sink component

Fields

port: usize

port to listen on

queue_size: usize

the size of the dispatch queue (also the batch size for Task store queue requests)

message_size: usize

size of an individual message chunk sent via zeromq (keep this small to avoid large RAM use, increase to reduce network bandwidth)

backend_address: String

address for the Task store postgres endpoint

metadata: WorkerMetadataSender

non-blocking handle to the background worker-metadata writer

Implementations

impl Sink

pub fn start( &self, services_arc: &Arc<ServiceCache>, sandboxes_arc: &Arc<SandboxCache>, progress_queue_arc: &Arc<InFlightSet>, done_tx: &SyncSender<TaskReport>, _job_limit: Option<usize>, dispatch_complete: &Arc<AtomicBool>, ) -> Result<(), Box<dyn Error>>

Starts a receiver/sink Server (ZMQ Pull), to accept processing responses. The sink shares state with other manager threads via queues for tasks in progress, as well as a queue for completed tasks pending persisting to disk. Termination is driven entirely by the shared dispatch_complete signal (set by the ventilator on a bounded run’s end or a graceful shutdown), not by _job_limit — the sink stops once dispatching is done and the in-flight set has drained.

Phase 5a (transport swap): the receive loop now runs on the pure-Rust async zeromq transport, driven by a current-thread tokio runtime this sink thread owns. zeromq delivers each result as one atomic multipart [ZmqMessage], which retires the libzmq frame-by-frame desync bug class (D-4/D-12): a malformed reply is just a message with too few frames, discarded by dropping it — it can no longer swallow the next reply. The blocking /data write + cortex.log parse stay fanned out to the unchanged phase-3 [run_writer] std-thread pool (size dispatcher.sink_writers); the blocking channel sends to it (and to the done channel) from this single-task runtime are the correct backpressure — when the disk or DB can’t keep up, the loop stops receiving and the workers back off. Note: zeromq exposes no TCP-keepalive knob, so dispatcher.tcp_keepalive_idle_seconds no longer applies to the sink PULL socket; keepalive was stability-only (remote-worker NAT mappings) and the lease reaper remains the correctness net.