Documentation
¶
Overview ¶
Package logstreamer implements the log-chunk publisher for sqi-worker.
A Publisher implements [executor.OutputHandler] and accumulates process output lines from stdout and stderr into protocol.LogChunkMsg batches, publishing them to the task.logs.<taskID> NATS JetStream subject.
Chunking ¶
Lines are buffered per-stream (stdout / stderr) per task attempt. A chunk is flushed immediately when:
- the buffered line count reaches Config.MaxLinesPerChunk, or
- the accumulated byte count reaches Config.MaxBytesPerChunk.
A background per-attempt goroutine also flushes buffered content at Config.FlushInterval intervals so the web UI log viewer feels live even for slowly printing processes.
Sequence numbers ¶
Each published chunk carries a monotonic protocol.LogChunkMsg.SeqNum starting at 1 for the first chunk of an attempt. Seq numbers increment across both stdout and stderr chunks so the server can replay all chunks in publication order regardless of NATS delivery order.
Lifecycle ¶
Call Publisher.FlushLogs after the process exits to drain all remaining buffered lines and stop the background flush goroutine. FlushLogs blocks until the drain is complete, providing the ordering guarantee that all log output arrives at the server before the terminal status message is published.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// MaxLinesPerChunk is the maximum number of lines batched into a single
// NATS message. A flush is triggered immediately when the buffer reaches
// this count. Default: 50.
MaxLinesPerChunk int
// MaxBytesPerChunk is the maximum total byte count of line content in a
// single NATS message. A flush is triggered when the accumulated bytes
// would meet or exceed this limit after adding a line. Default: 16384.
MaxBytesPerChunk int
// FlushInterval is the maximum time a line may sit in the buffer before
// being flushed regardless of chunk thresholds. Keeps the web UI log
// viewer feeling live. Default: 500 ms.
FlushInterval time.Duration
}
Config holds the tunable parameters for the log-chunk Publisher. Zero or negative values are replaced by the package defaults when New is called.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a Config populated with production-sensible defaults.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher implements executor.OutputHandler, accumulating process output lines from concurrent task goroutines into batched protocol.LogChunkMsg messages published to NATS JetStream.
Publisher is safe for concurrent use from multiple goroutines.
func New ¶
New returns a ready-to-use Publisher. Any zero or negative fields in cfg are replaced with defaults before use.
func (*Publisher) FlushLogs ¶
FlushLogs implements the executor.LogFlusher interface.
It signals the background flush goroutine to stop, waits for it to exit, then performs a final synchronous drain of any remaining buffered lines for both stdout and stderr. FlushLogs MUST be called after the task process has exited and before the terminal status message is published, providing the ordering guarantee that all log output is delivered to the server first.
Safety: the executor calls FlushLogs only after readWg.Wait() completes (all scanOutput goroutines — and thus all HandleLine calls — have returned). Because threshold-triggered publishes are synchronous inside HandleLine, there are no in-flight publishes for this attempt when FlushLogs runs. The only concurrent publisher is the background flushLoop goroutine, which is stopped via close(buf.done) + wg.Wait() before the final drain.
FlushLogs is idempotent: if no lines were ever received for the given attempt, or if FlushLogs has already been called, it returns nil immediately.
func (*Publisher) HandleLine ¶
HandleLine implements executor.OutputHandler.
It appends line to the in-memory buffer for (taskID, attemptID, stream). When the buffer meets the MaxLinesPerChunk or MaxBytesPerChunk threshold, a chunk is flushed immediately and synchronously — HandleLine does not return until the publish call completes. The per-attempt background goroutine handles the periodic FlushInterval drain between threshold events.
Concurrency note: threshold-triggered publishes happen synchronously inside HandleLine, so when all HandleLine callers return (i.e., when the executor's readWg.Wait() completes after process exit), no threshold flush is in flight. This is why [FlushLogs] can safely drain the remaining buffer without a concurrent write racing against it.
Source Files