feed

type Acker ¶

type Acker interface {
	Ack(ctx context.Context, eventID string) error
}

Acker is an optional interface a Source implements when its events need explicit acknowledgement after a successful local apply. The applier calls Ack(eventID) only after the SQLite tx that wrote applied_events + the data rows + the cursor advance commits — so a crash between fetch and ack causes the broker to redeliver, and the consumer-side applied_events guard makes the replay a no-op.

Sources that don't need per-event acks (e.g. the built-in file source, which reads from a directory and has no broker-side cursor) simply don't implement this interface. The applier type-asserts and falls back to a no-op when absent.

JetStream pull consumers in the service overlay are the primary implementor today (overlay/nats/controlfeed/source.go): they retain the per-message ack handle keyed by EventID until the applier confirms apply, then `AckSync` against the broker.

type Constructor ¶

type Constructor func(SourceConfig) (Source, error)

Constructor builds a Source from resolved config.

type Sink ¶

type Sink interface {
	// Append publishes one event to the feed. Input has EventID
	// populated (from outbox.event_id) and ControlVersion==0. On
	// success the returned event carries the broker-assigned
	// ControlVersion; on error the input is returned unchanged so the
	// pump's diagnostic logging has the originating event_id.
	Append(ctx context.Context, e controlevent.Event) (controlevent.Event, error)

	// Name is the backend identity (for logs).
	Name() string
}

Sink is the producer-side seam: a chassis writes one event into the fleet feed and gets back the same event stamped with its broker-assigned ControlVersion. Mirror of Source going the other direction — registered the same way (init + blank import), opened by name from --feed-sink.

The pump (chassis/controlpublish) drains the outbox and calls Append once per pending row. Implementations MUST use the event's EventID as their idempotent-publish key (Nats-Msg-Id on JetStream, filename on the file backend); retries with the same EventID must resolve to the same ControlVersion within the backend's dedup window. Beyond the dedup window, the consumer-side applied_events table is the load-bearing guard.

Append is single-call-per-event from the pump's point of view; an error sends the row back to the pending set for retry (with attempt_count + last_error bookkeeping). On success the pump writes published_control_version + published_at back to the outbox.

func OpenSink(name string, cfg SourceConfig) (Sink, error)

type SinkConstructor ¶

type SinkConstructor func(SourceConfig) (Sink, error)

SinkConstructor builds a Sink from resolved config. Same SourceConfig as the read side — backends that need richer config read their own env (the established seam convention; see chassis/continuation/factory.go and chassis/trace/factory.go).

type Source ¶

type Source interface {
	// Poll returns events with ControlVersion > sinceControlVersion,
	// sorted ascending by ControlVersion. An empty slice means "nothing
	// new" (not an error). Implementations must be cheap to call on a
	// timer and must not block beyond ctx.
	Poll(ctx context.Context, sinceControlVersion uint64) ([]controlevent.Event, error)

	// Name is the backend identity (for logs).
	Name() string
}

Source yields control events strictly after a cursor position.

func Open(name string, cfg SourceConfig) (Source, error)

type SourceConfig ¶

type SourceConfig struct {
	FileDir string
}

SourceConfig carries backend-selecting options resolved from chassis config. The file backend uses FileDir. Additional backends may extend this struct with their own fields; existing backends and callers are unaffected by added fields.