logstream

package
v0.1.16-rc1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 23, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrInvalidEventType = errors.New("logstream: invalid event type")

ErrInvalidEventType indicates an unknown SSE event type was provided.

View Source 
var ErrInvalidJobID = errors.New("logstream: invalid job ID (blank or whitespace)")

ErrInvalidJobID indicates a blank or whitespace-only job ID was provided.

View Source 
var ErrInvalidRunID = errors.New("logstream: invalid run ID (blank or whitespace)")

ErrInvalidRunID indicates a blank or whitespace-only run ID was provided.

View Source 
var ErrNoHub = errors.New("logstream: hub unavailable")

ErrNoHub indicates the hub is nil.

View Source 
var ErrStreamClosed = errors.New("logstream: stream closed")

ErrStreamClosed indicates the target stream is closed.

Functions

func Serve 

func Serve(w http.ResponseWriter, r *http.Request, hub *Hub, runID domaintypes.RunID, sinceID domaintypes.EventID) error

Serve streams events for the provided stream over SSE. sinceID must be a valid EventID (non-negative); callers should validate before calling. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func ServeFiltered 

func ServeFiltered(w http.ResponseWriter, r *http.Request, hub *Hub, runID domaintypes.RunID, sinceID domaintypes.EventID, filter func(Event) (Event, bool)) error

ServeFiltered streams events for the provided stream over SSE, applying an optional filter/transform function before writing frames.

If filter returns ok=false, the event is skipped. sinceID must be a valid EventID (non-negative); callers should validate before calling. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func ServeJob 

func ServeJob(w http.ResponseWriter, r *http.Request, hub *Hub, jobID domaintypes.JobID, sinceID domaintypes.EventID) error

ServeJob streams events for the provided job stream over SSE. sinceID must be a valid EventID (non-negative); callers should validate before calling. Returns ErrInvalidJobID if the job ID is blank or whitespace-only.

func WriteEventFrame 

func WriteEventFrame(w io.Writer, evt Event) error

WriteEventFrame writes a single SSE frame (id + event + data lines) to the writer.

Types

type Event 

type Event struct {
	ID   domaintypes.EventID
	Type domaintypes.SSEEventType
	Data []byte
}

Event represents a server-sent event frame produced by the hub.

type Hub 

type Hub struct {
	// contains filtered or unexported fields
}

Hub manages log streams published by nodes and consumed by SSE clients. Run-keyed streams carry lifecycle events (run, stage, done). Job-keyed streams carry container logs (log, done).

func NewHub 

func NewHub(opts Options) *Hub

NewHub constructs a log stream hub.

func (*Hub) Close 

func (h *Hub) Close(runID domaintypes.RunID)

Close tears down the stream and removes it from the hub. No-op if the run ID is blank.

func (*Hub) CloseAll 

func (h *Hub) CloseAll()

CloseAll tears down all streams (run and job) and clears the hub. Safe for graceful shutdown.

func (*Hub) CloseJob 

func (h *Hub) CloseJob(jobID domaintypes.JobID)

CloseJob tears down the job stream and removes it from the hub.

func (*Hub) Ensure 

func (h *Hub) Ensure(runID domaintypes.RunID) error

Ensure creates the stream if it does not already exist. Returns an error if the run ID is blank or whitespace-only.

func (*Hub) EnsureJob 

func (h *Hub) EnsureJob(jobID domaintypes.JobID) error

EnsureJob creates the job stream if it does not already exist. Returns ErrInvalidJobID if the job ID is blank or whitespace-only.

func (*Hub) PublishJobLog 

func (h *Hub) PublishJobLog(ctx context.Context, jobID domaintypes.JobID, record LogRecord) error

PublishJobLog appends a log record to a job stream. Returns ErrInvalidJobID if the job ID is blank or whitespace-only.

func (*Hub) PublishJobRetention 

func (h *Hub) PublishJobRetention(ctx context.Context, jobID domaintypes.JobID, hint RetentionHint) error

PublishJobRetention appends a retention hint to a job stream. Returns ErrInvalidJobID if the job ID is blank or whitespace-only.

func (*Hub) PublishJobStatus 

func (h *Hub) PublishJobStatus(ctx context.Context, jobID domaintypes.JobID, status Status) error

PublishJobStatus appends a terminal status event and closes the job stream. Returns ErrInvalidJobID if the job ID is blank or whitespace-only.

func (*Hub) PublishLog 

func (h *Hub) PublishLog(ctx context.Context, runID domaintypes.RunID, record LogRecord) error

PublishLog appends a log record to a stream. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func (*Hub) PublishRetention 

func (h *Hub) PublishRetention(ctx context.Context, runID domaintypes.RunID, hint RetentionHint) error

PublishRetention appends a retention hint to a stream. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func (*Hub) PublishRun 

func (h *Hub) PublishRun(ctx context.Context, runID domaintypes.RunID, run api.RunSummary) error

PublishRun appends a typed run snapshot to a stream.

The payload is strongly typed as api.RunSummary to prevent accidental publication of non‑JSON payloads (e.g., raw []byte or strings). The hub still performs generic JSON marshaling internally, but this boundary keeps the "run" event contract consistent and JSON‑serializable.

Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func (*Hub) PublishStage 

func (h *Hub) PublishStage(ctx context.Context, runID domaintypes.RunID, record LogRecord) error

PublishStage appends a stage progress event to a run stream. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func (*Hub) PublishStatus 

func (h *Hub) PublishStatus(ctx context.Context, runID domaintypes.RunID, status Status) error

PublishStatus appends a terminal status event and closes the stream. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func (*Hub) Snapshot 

func (h *Hub) Snapshot(runID domaintypes.RunID) []Event

Snapshot returns a copy of buffered events for the stream. Returns nil if the run ID is blank or the stream does not exist.

func (*Hub) SnapshotJob 

func (h *Hub) SnapshotJob(jobID domaintypes.JobID) []Event

SnapshotJob returns a copy of buffered events for the job stream. Returns nil if the job ID is blank or the stream does not exist.

func (*Hub) Subscribe 

func (h *Hub) Subscribe(ctx context.Context, runID domaintypes.RunID, sinceID domaintypes.EventID) (Subscription, error)

Subscribe registers a consumer for the stream starting after the provided id. The sinceID must be a valid EventID (non-negative); invalid IDs are rejected. Returns ErrInvalidRunID if the run ID is blank or whitespace-only.

func (*Hub) SubscribeJob 

func (h *Hub) SubscribeJob(ctx context.Context, jobID domaintypes.JobID, sinceID domaintypes.EventID) (Subscription, error)

SubscribeJob registers a consumer for the job stream starting after the provided id. Returns ErrInvalidJobID if the job ID is blank or whitespace-only.

type LogRecord 

type LogRecord struct {
	Timestamp string `json:"timestamp"`
	Stream    string `json:"stream"`
	Line      string `json:"line"`

	// NodeID identifies the execution node that produced this log line (NanoID-backed).
	// Empty when the source is not node-bound (e.g., hub-generated events).
	NodeID domaintypes.NodeID `json:"node_id,omitempty"`

	// JobID is the identifier of the job that produced this log line (KSUID-backed).
	// Empty for events not tied to a specific job.
	JobID domaintypes.JobID `json:"job_id,omitempty"`

	// JobType indicates the job phase type (e.g., "pre_gate", "mig", "post_gate").
	// Empty when not applicable or unknown. Uses domain type for type-safe identification.
	JobType domaintypes.JobType `json:"job_type,omitempty"`
}

LogRecord represents a structured log frame. The enriched fields (NodeID, JobID, JobType) provide execution context so clients can correlate log lines with specific nodes, jobs, and Migs pipeline stages. These fields are optional — older or internal-only log sources may omit them. Uses domain types (NodeID, JobID, JobType) for type-safe identification.

type Options 

type Options struct {
	// BufferSize controls the per-subscriber channel size (default: 32).
	BufferSize int
	// HistorySize bounds the number of events retained for resumption (default: 256, min: BufferSize).
	HistorySize int
}

Options configures the hub.

type RetentionHint 

type RetentionHint struct {
	Retained bool            `json:"retained"`
	TTL      string          `json:"ttl"`
	Expires  string          `json:"expires_at"`
	Bundle   domaintypes.CID `json:"bundle_cid,omitempty"`
}

RetentionHint carries retention metadata emitted on the stream.

type Status 

type Status struct {
	Status string `json:"status"`
}

Status announces terminal stream states.

type Subscription 

type Subscription struct {
	Events <-chan Event
	// contains filtered or unexported fields
}

Subscription delivers events to a consumer.

func (Subscription) Cancel 

func (s Subscription) Cancel()

Cancel terminates the subscription.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.