compactor

Oct 17, 2026

Changes in this version

+ const DefaultTelemetryKept

type Metrics

+ WebhookCompactionRun func(ctx context.Context, source string, duration, lag time.Duration, ...)

type Tuning

+ TelemetryKept time.Duration

type WindowStore

+ CompactedDigest func(ctx context.Context, h whstore.Window) (digest string, current bool, err error)

+ RecordCompactionRun func(ctx context.Context, run whstore.CompactionRun) error

+ RollupTelemetry func(ctx context.Context, before time.Time) error

Oct 16, 2026

Changes in this version

type Tables

+ ReplaceView func(ctx context.Context, tg whtable.Target, src whsource.Source) error

type Worker

+ func (w *Worker) ReplaceViews(ctx context.Context) error

Oct 16, 2026

Changes in this version

+ const DefaultRollupCheckEvery

+ const DefaultRollupDeleteDelay

type Objects

+ PutObject func(ctx context.Context, bucket, key string, data []byte, contentType string) error

type Tables

+ RegisterRawWindow func(ctx context.Context, tg whtable.Target, src whsource.Source, window time.Time) error

+ UnregisterRawWindow func(ctx context.Context, tg whtable.Target, src whsource.Source, window time.Time) error

type Tuning

+ RollupCheckEvery time.Duration

+ RollupDeleteDelay time.Duration

type WindowStore

+ ClaimRollup func(ctx context.Context, source string, start time.Time, lease time.Duration) (bool, error)

+ RawRegistrable func(ctx context.Context, source string, since time.Time) ([]whstore.Window, error)

type Worker

+ func (w *Worker) Rollup(ctx context.Context) error

Sep 27, 2026

Changes in this version

+ const DefaultExpiredWindowsKept

type Tuning

+ ExpiredWindowsKept time.Duration

type WindowStore

+ DeleteExpired func(ctx context.Context, before time.Time) (int64, error)

Sep 26, 2026

Changes in this version

+ const DefaultBatch

+ const DefaultGrace

+ const DefaultLease

+ const DefaultPoll

+ const DefaultRetentionEvery

+ const DefaultRetryBackoff

+ const ResultCompacted

+ const ResultFailed

+ type Deps struct

+ Bucket string

+ Logger *slog.Logger

+ Metrics Metrics

+ Now func() time.Time

+ Objects Objects

+ Resources WindowResources

+ Sources Sources

+ Tables Tables

+ Windows WindowStore

+ type Metrics interface

+ WebhookCompaction func(ctx context.Context, source, result string)

+ WebhookDuplicatesDropped func(ctx context.Context, source string, n int64)

+ type Objects interface

+ DeleteObject func(ctx context.Context, bucket, key string) error

+ GetObject func(ctx context.Context, bucket, key string) ([]byte, error)

+ ListKeys func(ctx context.Context, bucket, prefix string) ([]string, error)

+ type Sources interface

+ Get func(ctx context.Context, name string) (whsource.Source, error)

+ List func(ctx context.Context) ([]whsource.Source, error)

+ type StoredWindow struct

+ Key string

+ ResourceID string

+ type Tables interface

+ RegisterWindow func(ctx context.Context, tg whtable.Target, src whsource.Source, window time.Time, ...) error

+ S3Location func(prefix string) string

+ SyncRaw func(ctx context.Context, tg whtable.Target, src whsource.Source) error

+ TargetFor func(connection string) (whtable.Target, error)

+ UnregisterWindow func(ctx context.Context, tg whtable.Target, src whsource.Source, window time.Time) error

+ type Tuning struct

+ Batch int

+ Grace time.Duration

+ Lease time.Duration

+ Poll time.Duration

+ RetentionEvery time.Duration

+ RetryBackoff time.Duration

+ type WindowResources interface

+ Delete func(ctx context.Context, id string) error

+ Key func(ctx context.Context, id string) (string, bool, error)

+ Put func(ctx context.Context, src whsource.Source, window time.Time, existingID string, ...) (StoredWindow, error)

+ type WindowStore interface

+ ClaimOwed func(ctx context.Context, endedBy time.Time, lease time.Duration, limit int) ([]whstore.Window, error)

+ Expirable func(ctx context.Context, source string, cutoff time.Time) ([]whstore.Window, error)

+ MarkExpired func(ctx context.Context, h whstore.Window) error

+ MarkRawDeleted func(ctx context.Context, h whstore.Window) error

+ MarkUnregistered func(ctx context.Context, h whstore.Window) error

+ PruneCounts func(ctx context.Context, before time.Time) error

+ RawDeletable func(ctx context.Context, source string, cutoff time.Time) ([]whstore.Window, error)

+ RecordCompacted func(ctx context.Context, h whstore.Window, c whstore.Compaction) error

+ RecordFailure func(ctx context.Context, h whstore.Window, reason string, hold time.Duration) error

+ RecordLocation func(ctx context.Context, h whstore.Window, resourceID, location string) error

+ type Worker struct

+ func New(t Tuning, d Deps) *Worker

+ func (w *Worker) Pass(ctx context.Context) bool

+ func (w *Worker) Retention(ctx context.Context)

+ func (w *Worker) Start(ctx context.Context)

+ func (w *Worker) Stop()