Documentation
¶
Index ¶
- Constants
- func EncodeCompactRow(orderedCols []discovery.Column, record map[string]any) (json.RawMessage, error)
- func StartIngestWorker(ctx context.Context, queue Queue, cache cache.Cache, ...) (stop func(context.Context) error, failed <-chan error, err error)
- type ClaimConfig
- type EventMessage
- type IngestWorker
- type Queue
- type Sweeper
Constants ¶
const BufferConsumerName = "buffer-consumer"
BufferConsumerName is the durable consumer name used by the ingest worker. The Active Sweeper references this to read the AckFloor.
const FormatJSONCompactEachRow = "JSONCompactEachRow"
FormatJSONCompactEachRow is the only row format the envelope carries today. It is stated on the wire rather than assumed so a reader can tell an envelope it understands from one it doesn't, and so a second format can be added without a third guess at what the bytes mean.
Variables ¶
This section is empty.
Functions ¶
func EncodeCompactRow ¶ added in v0.2.0
func EncodeCompactRow(orderedCols []discovery.Column, record map[string]any) (json.RawMessage, error)
EncodeCompactRow renders one record as a single JSONCompactEachRow line: a JSON array carrying exactly one value per INSERTABLE column, in declaration order (discovery orders TableSchema.Columns by system.columns.position). A column the record does not carry encodes as null. For a NON-nullable column with a default the insert turns that back into the default (input_format_null_as_default); on a NULLABLE column ClickHouse stores the NULL, because only an ABSENT key ever took the default and a positional row cannot express absence. worker.go pins input_format_null_as_default=1 explicitly, alongside date_time_input_format, so a server-default change cannot silently alter either. The result has no trailing newline — the caller joins lines.
This is serialization ONLY. It performs no validation and makes no decision about a value: schema validation upstream has already rejected unknown keys and unacceptable types, so there is nothing here to reject and nothing to coerce. Record values arrive json.Number-preserving (every decoder on the ingest path sets UseNumber), and json.Marshal writes a json.Number as its exact digits, so a 64-bit id past 2^53 keeps every one of them.
Transitional: replaced by chtypes RowsExport; serialization only, never add rules here.
func StartIngestWorker ¶
func StartIngestWorker( ctx context.Context, queue Queue, cache cache.Cache, target func(tenant.ID) chconn.Target, dlqEnabled func(id tenant.ID, table string) bool, ) (stop func(context.Context) error, failed <-chan error, err error)
StartIngestWorker starts the batch consumer. stop drains it under the given deadline. failed receives at most one error, if the worker ends on its own: ingestion has stopped and nothing inside the worker can bring it back, so the caller must treat it as fatal (internal/app returns it from Run, which stops the process — the next boot recreates the consumer). The worker has already flushed and acked what it held by the time failed fires; stop is still the caller's to call.
Types ¶
type ClaimConfig ¶ added in v0.2.0
type ClaimConfig struct {
// Every is how often membership is read and units claimed or given up;
// default coord.RetryPeriod.
Every time.Duration
// Handover bounds how long a unit given up waits for its delivered rows
// to settle before it is released; default the worker's ack_wait, after
// which the broker redelivers them anyway.
Handover time.Duration
// MaxHeld is the process's budget of rows delivered and not yet settled,
// shared out per unit (see Memory above); default the worker's
// maxAckPending.
MaxHeld int
// UnownedEvery is how often the lowest live member reads how many units
// have rows and no owner; default 15s.
UnownedEvery time.Duration
// SlotExpiry frees a delivered row's MaxHeld slot if the worker never
// settles it: by then the broker redelivers the row anyway. Default the
// worker's ack_wait.
SlotExpiry time.Duration
}
ClaimConfig tunes ClaimShards.
type EventMessage ¶
type EventMessage struct {
TableName string `json:"table_name"`
Scope string `json:"scope"`
ReceivedTimestamp string `json:"received_timestamp"`
Format string `json:"format"`
Columns []string `json:"columns"`
Row json.RawMessage `json:"row"`
}
EventMessage is the wire format published to the MQ.
Row data travels POSITIONALLY: Row is one JSONCompactEachRow line (a JSON array, no trailing newline) and Columns names its positions in the table's declaration order. The two are only meaningful together — a reader that cannot pair them (a length mismatch, an undecodable row, a repeated column name) has no way to map a value to a column and must fail closed rather than guess.
type IngestWorker ¶
type IngestWorker struct {
// contains filtered or unexported fields
}
type Queue ¶ added in v0.2.0
type Queue interface {
mq.ConsumerManager
mq.DeadLetterer
}
Queue is what the worker needs from the MQ: a durable consumer on the ingest queue and somewhere to park what it cannot write. mq.Broker satisfies it.
func ClaimShards ¶ added in v0.2.0
func ClaimShards(q Queue, c coord.Coordinator, cfg ClaimConfig) (Queue, error)
ClaimShards wraps q so that its consumer takes only the units this process is assigned through c, which every ingest process must share and which must implement coord.Observer. q must implement mq.Sharded.
type Sweeper ¶
type Sweeper struct {
// contains filtered or unexported fields
}
Sweeper implements the Active Sweeper pattern. It runs every minute and asks the MQ to purge the ingest events that satisfy BOTH conditions:
- ACKed by the buffer consumer (written to ClickHouse)
- Older than their tenant's gap window (no longer needed for SSE replay)
This guarantees: healthy state keeps exactly each tenant's gap_window of rolling data; ClickHouse down freezes purging; a catastrophic outage fills a tenant's queue to its byte budget and triggers backpressure (mq.ErrQueueFull). How the MQ finds the purge point is its own business (see mq.Purger).
func NewSweeper ¶
NewSweeper creates the Active Sweeper. gapWindows is resolved per sweep. One runs per queue: internal/app starts it under the coord sweeper lease.
Source Files