capturelog

package
v0.22.0 Latest Latest
Warning

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

Go to latest
Published: Nov 11, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package capturelog is the shared persistence engine for append-only HTTP capture tables: the single writer per process that drains a recorder, batching, retries, availability checks, loss logging, retention by age and row cap under an advisory lock, the bounded viewer read gate and the prune command. It knows nothing about HTTP disclosure; each table supplies its typed declaration and a redact function.

Index

Constants

View Source
const (

	// PruneBatch is the most rows one retention transaction removes.
	PruneBatch = 1_000
	// PruneBatches bounds one scheduled pass.
	PruneBatches = 50
)

Variables

View Source
var ErrRetentionBusy = safeerror.NewFault(fault.Conflict, "another process holds the capture retention lock")

ErrRetentionBusy reports that another process holds this table's retention lock for the schema; the batch was skipped.

Functions

func Command

func Command(name cli.Name, summary string, construct func(foundation.Resolver) (Pruner, error)) (cli.Declaration, error)

Command declares `<name> prune` for the ordinary CLI registry. construct resolves a read-and-prune log after boot; it works while capture is disabled, as long as the table's database and schema are configured.

func DroppedBefore

func DroppedBefore(count uint64) int64

DroppedBefore clamps a loss count into its signed column.

Types

type Dependencies

type Dependencies struct {
	DB *database.DB
	// Schema holds the table; it must be a valid identifier.
	Schema string
	// Clock supplies retention cutoffs. Capture times come from the producer.
	Clock  clock.Clock
	Logger *slog.Logger
}

Dependencies are borrowed; the log never closes them.

type Engine

type Engine interface {
	Pruner
	Run(ctx context.Context, roles Roles) error
	Close(ctx context.Context) error
	Done() <-chan struct{}
	Check(ctx context.Context)
	Flush(ctx context.Context)
	PrunePass(ctx context.Context) bool
	Read(ctx context.Context, fn func(context.Context, *database.Tx) error) error
	Transaction(ctx context.Context, readOnly bool, fn func(*database.Tx) error) error
	Lock(ctx context.Context, tx *database.Tx) (bool, error)
}

Engine is a log's operations without its row and snapshot types. Public capture stores hold it through this interface, so packages that only name a store never instantiate the generic engine and its table declaration.

type Limits

type Limits struct {
	Retention time.Duration
	MaxRows   int
}

Limits bound the table. Each public capture config validates its own range.

type Log

type Log[S, M any] struct {
	// contains filtered or unexported fields
}

Log is one table's writer, retention and reads. A nil recorder gives a log that reads and prunes but never writes.

func New

func New[S, M any](table Table[M], recorder *httpcapture.Recorder[S], redact Redact[S, M], deps Dependencies, limits Limits) (*Log[S, M], error)

New validates its inputs without I/O.

func (*Log[S, M]) Check

func (l *Log[S, M]) Check(ctx context.Context)

Check queries the table's availability now and logs a change.

func (*Log[S, M]) Close

func (l *Log[S, M]) Close(ctx context.Context) error

Close stops admission, writes what remains within a short drain budget and counts the rest as lost. It runs after every producer has drained.

func (*Log[S, M]) Cutoff

func (l *Log[S, M]) Cutoff() (temporal.DateTime, error)

Cutoff is now minus retention; reads hide older rows before pruning does.

func (*Log[S, M]) Done

func (l *Log[S, M]) Done() <-chan struct{}

Done closes once Close has finished.

func (*Log[S, M]) Estimate

func (l *Log[S, M]) Estimate(ctx context.Context, cutoff temporal.DateTime, capped bool) (int64, error)

Estimate counts rows a prune would remove without deleting them.

func (*Log[S, M]) Flush

func (l *Log[S, M]) Flush(ctx context.Context)

Flush writes every queued capture now, checking the table first when it has not been checked yet.

func (*Log[S, M]) Format

func (*Log[S, M]) Format(s fmt.State, _ rune)

Format writes a fixed label without exposing dependencies or captures.

func (*Log[S, M]) Label

func (l *Log[S, M]) Label() string

Label names the table's rows in logs and command output.

func (*Log[S, M]) Lock

func (l *Log[S, M]) Lock(ctx context.Context, tx *database.Tx) (bool, error)

Lock takes the transaction-scoped retention lock for this table and schema without waiting; processes that lose the race skip the batch.

func (*Log[S, M]) Now

func (l *Log[S, M]) Now() (temporal.DateTime, error)

Now is the log clock's current instant.

func (*Log[S, M]) Prune

func (l *Log[S, M]) Prune(ctx context.Context, cutoff temporal.DateTime, capped bool, batch, maxBatches int) (int64, bool, error)

Prune deletes rows captured before cutoff, then, when capped, rows beyond MaxRows (per partition value when the table has one), oldest first. Every batch is one transaction holding the transaction-scoped retention lock; a process that loses it stops with ErrRetentionBusy. It reports whether it stopped at maxBatches.

func (*Log[S, M]) PrunePass

func (l *Log[S, M]) PrunePass(ctx context.Context) bool

PrunePass runs one scheduled pass and reports whether it stopped at its batch bound, so the next pass comes sooner. The first pass also checks the table, which logs once when it is not migrated.

func (*Log[S, M]) Read

func (l *Log[S, M]) Read(ctx context.Context, fn func(context.Context, *database.Tx) error) error

Read admits at most two concurrent viewer reads, each bounded by two seconds including the wait for a slot, in a read-only transaction.

func (*Log[S, M]) Run

func (l *Log[S, M]) Run(ctx context.Context, roles Roles) error

Run writes captures and prunes the table until ctx ends. Database trouble never stops it: an unavailable table degrades capture with one logged reason, a failed batch is retried twice and then row by row, and only the rows that still fail are counted as lost.

func (*Log[S, M]) Transaction

func (l *Log[S, M]) Transaction(ctx context.Context, readOnly bool, fn func(*database.Tx) error) error

Transaction scopes fn to the capture schema inside one owned transaction on the borrowed primary pool. Reads use the primary too: recent captures may not have reached a replica yet.

type PruneCommand

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

PruneCommand is a validated prune operation. Without --before it applies the configured retention and row cap; with it, only the explicit cutoff, which must not be later than now. Without --apply it only counts.

func ParsePrune

func ParsePrune(name cli.Name, args []string, help io.Writer) (PruneCommand, error)

ParsePrune parses prune arguments without database I/O. It writes help on request and then returns flag.ErrHelp.

func (PruneCommand) Run

func (c PruneCommand) Run(ctx context.Context, target Pruner, output io.Writer) error

Run counts the matching captures or, with --apply, removes them in bounded transactions and reports the committed count, also when a later batch fails. It never prints captured data.

type Pruner

type Pruner interface {
	Label() string
	Now() (temporal.DateTime, error)
	Cutoff() (temporal.DateTime, error)
	Estimate(ctx context.Context, cutoff temporal.DateTime, capped bool) (int64, error)
	Prune(ctx context.Context, cutoff temporal.DateTime, capped bool, batch, maxBatches int) (int64, bool, error)
}

Pruner is the retention surface the prune command drives. Every Log implements it.

type Redact

type Redact[S, M any] func(snapshot S, droppedBefore uint64) (M, error)

Redact turns one drained snapshot and the loss its lane carried into a row.

type Roles

type Roles struct {
	// Write drains the recorder. The table is checked on the first record.
	Write bool
	// Retain prunes on a timer; its first pass also checks the table.
	Retain bool
}

Roles chooses what Run does in this process.

type Table

type Table[M any] struct {
	// Name is the table name used by the availability check.
	Name string
	// Label names the rows in logs and command output, e.g. "webhook capture".
	Label string
	// Lock is the advisory-lock namespace retention holds per schema.
	Lock  string
	Query func() orm.Query[M]
	ID    orm.Key[M, model.ID[M]]
	At    orm.Time[M, temporal.DateTime]
	// Partition, when set, applies the row cap per distinct value of this
	// column instead of to the whole table.
	Partition *orm.Text[M, string]
	Insert    func(context.Context, *database.Tx, []M) error
	// Size estimates a row's payload for write batching.
	Size func(M) int
}

Table describes one capture table through its handwritten declaration.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL