sink

Arrakis sink element.

ArrakisSink dataclass

ArrakisSink(*, channels, publisher_id, replay_id=None, summary_interval=900.0, backlog_warn=10, max_publish_failures=10, publish_timeout=30.0)

Bases: TSSink

Sink element that streams channel data to Arrakis.

Parameters:

Name	Type	Description	Default
channels	list[str]	Channels to publish to Arrakis. Sink pads are automatically generated for each channel, with the channel name as the pad name.	required
publisher_id	str	admin-assigned publisher ID	required
replay_id	str or None	optional replay namespace identifier	None
summary_interval	float	Seconds between INFO-level activity summaries reporting blocks queued, the newest block's lag behind real time, and the publish backlog; its steady presence doubles as a liveness heartbeat. 0 disables.	900
backlog_warn	int	Warn when more than this many blocks are waiting in the publish queue, i.e. publishing is not keeping up; recovery is logged once the backlog drops below half the threshold.	10
max_publish_failures	int	Abort the pipeline after this many consecutive blocks fail to publish (:class:arrakis.PublishError, or delivery not confirmed within publish_timeout); until then each failed block is logged at ERROR and dropped, so a single unwritable partition degrades that partition's data instead of crash-looping the whole pipeline and losing every channel's queued backlog on each restart. A successful publish resets the count. 0 disables the cap.	10
publish_timeout	float	Seconds to wait for a block's delivery to be confirmed before counting it as a failure. Without it a dead broker stalls each block for the producer's own message timeout (5 minutes by default). 0 waits indefinitely.	30

Notes

Logging: At INFO the sink logs the publisher it registers, the first block queued, backlog and recovery transitions, and the periodic activity summary; per-block detail is DEBUG. Blocks dropped on publish failure are logged at ERROR.

Shutdown: At EOS the publish thread drains the queued backlog, unless publishing is failing, in which case the backlog is dropped (logged at ERROR with its size) so a dead broker cannot hold the pipeline open. The thread is a daemon: a failure elsewhere in the pipeline never leaves the process waiting on it.

Thread safety: Marked thread_safe = True. Pad layout: N sink pads (one per channel, no source pads). The sink pads' pull callbacks CAN run concurrently in the same wave; internal (which calls process) runs alone.

``pull`` is the inherited per-pad-keyed
``TimeSeriesMixin.pull``. All element-wide mutation --
starting/stopping the publisher thread, ``self.queue.put``,
checking ``exception_event``, the logging counters -- happens
in ``process``, which runs alone. Communication with the
background publish thread already uses thread-safe primitives
(``queue.Queue``, ``threading.Event``); its ``self.at_eos``
read is a plain attribute read, safe from another thread.

Speedup from threading is modest (publishing is already
offloaded to the background thread), but marking avoids
forcing a serialization point in the graph wave against
neighboring thread_safe elements.

**Future editors MUST preserve thread safety**: do not
override ``pull`` with element-wide mutation; keep
publisher/queue/thread state mutations confined to ``process``
and the background thread.

static_sink_pads property

static_sink_pads

Define sink pads from channels.

configure

configure()

Configure the ArrakisSink.

Source code in sgn_arrakis/sink.py

def configure(self) -> None:
    """Configure the ArrakisSink."""
    # enable the adapter; stride/align_to are set in on_startup()
    # once the publisher's expected stride is known.
    self.adapter_config.alignment()

    self.publisher: Publisher = Publisher(
        self.publisher_id, replay_id=self.replay_id
    )
    self.queue: queue.Queue = queue.Queue()
    self.exception_event: threading.Event = threading.Event()
    self.stop_event: threading.Event = threading.Event()
    # publish in a background thread so that we don't eat time
    # from the main sgn graph execution (makes internal pad
    # execution ~10x faster); a daemon, so a pipeline that fails
    # elsewhere (EOS never reaches this sink) can still exit
    self.thread: threading.Thread = threading.Thread(
        target=self._arrakis_publish, daemon=True
    )

    # Logging state: per-cadence summaries and backlog transitions
    self._blocks_queued: int = 0
    self._backlogged: bool = False
    self._last_summary: float = time.monotonic()
    self._summary_blocks: int = 0
    self._last_block_end_ns: int | None = None

on_startup

on_startup()

Register the publisher and align to its expected block stride.

Source code in sgn_arrakis/sink.py

def on_startup(self) -> None:
    """Register the publisher and align to its expected block stride."""
    self.publisher.register()
    invalid_chans = set(self.sink_pad_names) - set(self.publisher.channels.keys())
    if invalid_chans:
        msg = f"invalid publisher channels: {invalid_chans}"
        raise ValueError(msg)
    assert self.publisher.stride is not None
    stride = Offset.fromns(self.publisher.stride)
    self.adapter_config.alignment(stride=stride, align_to=stride)
    self.logger.info(
        "Publishing %d channel(s) to Arrakis as publisher %s%s (stride %g s)",
        len(self.sink_pad_names),
        self.publisher_id,
        f", replay {self.replay_id}" if self.replay_id else "",
        self.publisher.stride / Time.SECONDS,
    )

process

process(input_frames)

Process input frames and publish to Arrakis.

Parameters:

Name	Type	Description	Default
input_frames	dict[SinkPad, TSFrame]	mapping of sink pads to their input frames	required

Source code in sgn_arrakis/sink.py

def process(self, input_frames: dict[SinkPad, TSFrame]) -> None:
    """Process input frames and publish to Arrakis.

    Parameters
    ----------
    input_frames : dict[SinkPad, TSFrame]
        mapping of sink pads to their input frames
    """
    try:
        self._process(input_frames)
    except BaseException:
        # Ensure the background publish thread is stopped so the
        # pipeline can unwind cleanly rather than hanging on a
        # live non-daemon thread.
        self.stop()
        raise

stop

stop()

Stop the publish thread promptly, without draining the queue.

Source code in sgn_arrakis/sink.py

def stop(self):
    """Stop the publish thread promptly, without draining the queue."""
    self.stop_event.set()
    self._join()