controlpublish

Documentation

Overview

Package controlpublish is the producer half of fleet sync: a background pump that drains the control_events_outbox table and publishes pending rows via feed.Sink. The lifecycle mirrors chassis/controlapply.Controller (Start/Stop, ticker, ctx.Done, errors logged not fatal). It is inert unless --feed-sink != nop, so single-node behaviour is unchanged.

Producer obligations (see contract §5 + the overlay-repo design doc todo-fleet-sync-producer.md):

• Admin handlers upload the artifact bytes to the artifact store BEFORE opening the local DB tx.
• In the tx, the handler appends a row to control_events_outbox carrying event_id (UUID), event_type, the decomposed fields, and the full canonical payload as payload_json.
• The handler commits.
• This pump picks the row up asynchronously, publishes via Sink.Append, and writes the broker-assigned ControlVersion back to the row on success.

Crash safety: anything in the outbox WILL be published once the broker is reachable. Anything not in the outbox was never accepted. Retries reuse the same event_id, so backends with idempotent publish semantics (JetStream Nats-Msg-Id, the file Sink's filename-as-key) resolve duplicates naturally; the consumer-side applied_events table is the load-bearing semantic dedup.

Index

• func AppendOutbox(ctx context.Context, tx *sql.Tx, eventID, eventType, tenantID, stackID string, ...) error
• type Controller
  • func NewController(ctx context.Context, pu *processor.Unit, sink feed.Sink) *Controller
  • func (c *Controller) DrainForTest(ctx context.Context)
  • func (c *Controller) Start()
  • func (c *Controller) Stop()

Functions

func AppendOutbox

func AppendOutbox(
	ctx context.Context,
	tx *sql.Tx,
	eventID, eventType, tenantID, stackID string,
	version, baseVersion int64,
	artifactRef, checksum string,
	payloadJSON []byte,
) error

AppendOutbox is the helper admin handlers call inside their existing transactions. Generating event_id is the caller's responsibility (chassis/hxid.New is the convention) so callers can also return the event_id back to the user (audit log / debugging surface) if they want.

payloadJSON is the canonical Event JSON minus ControlVersion (the Sink stamps that on publish). Keeping the full doc in the blob means new fields land without column migrations.

Types

type Controller

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

Controller is the background pump. Lifecycle mirrors controlapply.Controller.

func NewController

func NewController(ctx context.Context, pu *processor.Unit, sink feed.Sink) *Controller

NewController returns a pump bound to the chassis unit. sink may be nil when feed-sink=nop; enabled() returns false in that case and Start/Stop are no-ops.

func (*Controller) DrainForTest

func (c *Controller) DrainForTest(ctx context.Context)

DrainForTest exposes the internal drain loop for integration tests that drive the pump synchronously instead of relying on the ticker. Not part of the public API; do not call from production.

func (*Controller) Start

func (c *Controller) Start()

Start launches the pump goroutine. No-op when disabled.

func (*Controller) Stop

func (c *Controller) Stop()

Stop signals shutdown and waits for the pump to drain in-flight publishes. No-op when disabled.