whstore

package
v1.136.0 Latest Latest
Warning

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

Go to latest
Published: Sep 26, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Overview

Package whstore is the control data of inbound webhooks (#1870): which compaction windows of each source have segments, which of them are compacted and where, what retention has done to them, and the request counts and rejections a source's page reports. Events are never written here.

Index

Constants

This section is empty.

Variables

View Source
var ErrNotFound = errors.New("webhook window not found")

ErrNotFound is returned when a write names a window no row holds.

Functions

This section is empty.

Types

type Compaction

type Compaction struct {
	Segments   int
	Events     int64
	Duplicates int64
	Digest     string
	ResourceID string
	Location   string
}

Compaction is what one compaction of a window produced.

type Count

type Count struct {
	Source  string
	Minute  time.Time
	Outcome string
	Count   int64
}

Count is a number of requests of one source with one outcome in one minute.

type Rejection

type Rejection struct {
	Source  string    `json:"-"`
	At      time.Time `json:"at"`
	Outcome string    `json:"outcome"`
	Reason  string    `json:"reason"`
}

Rejection is one refused request: when, which outcome, and why. Never the body.

type Status

type Status struct {
	// LastHour and LastDay are request counts by outcome.
	LastHour map[string]int64 `json:"last_hour"`
	LastDay  map[string]int64 `json:"last_day"`
	// LastSegmentAt is when a segment of accepted events was last written,
	// which is when the source last received an event. A sender that stops
	// produces no error on the receiving side; this is how that shows.
	LastSegmentAt *time.Time `json:"last_segment_at"`
	// LastCompactedWindow is the start of the newest window whose Parquet
	// file holds every event its segments do.
	LastCompactedWindow *time.Time `json:"last_compacted_window"`
	// Pending counts windows owed a compaction: not yet compacted, or
	// dirtied by a segment written after they were.
	Pending int `json:"pending"`
	// Failing counts pending windows whose last compaction attempt failed.
	Failing   int    `json:"failing"`
	LastError string `json:"last_error,omitempty"`
	// OldestWindow is the start of the oldest window still held.
	OldestWindow *time.Time  `json:"oldest_window"`
	Rejections   []Rejection `json:"rejections"`
}

Status is what a source's page reports about it.

type Store

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

Store reads and writes the webhook control tables.

func New

func New(db *sql.DB) *Store

New creates a store over the platform database.

func (*Store) ClaimOwed

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

ClaimOwed claims up to limit windows owed a compaction that ended at or before endedBy, for lease. A window another replica holds is skipped until its lease runs out, and a window that failed is held back until the backoff recorded against it passes.

func (*Store) Expirable

func (s *Store) Expirable(ctx context.Context, source string, cutoff time.Time) ([]Window, error)

Expirable returns the windows of source that ended at or before cutoff and are not yet expired, and the expired ones a segment has landed in since, whose raw objects have to be deleted again so the view does not serve them.

func (*Store) MarkExpired

func (s *Store) MarkExpired(ctx context.Context, w Window) error

MarkExpired records that a window is gone: its partition, its resource and its raw segments.

func (*Store) MarkRawDeleted

func (s *Store) MarkRawDeleted(ctx context.Context, w Window) error

MarkRawDeleted records that a window's raw segments were deleted.

func (*Store) MarkSegment

func (s *Store) MarkSegment(ctx context.Context, source string, start time.Time, length time.Duration) error

MarkSegment records that a segment for the window starting at start and length long was written. A window already compacted becomes owed another compaction, because its Parquet file no longer holds every event its segments do. A window recorded with a shorter length is lengthened: a source whose compact_every was raised can land a segment in a window the old setting had already ended.

func (*Store) MarkUnregistered

func (s *Store) MarkUnregistered(ctx context.Context, w Window) error

MarkUnregistered records that a window's compacted partition was removed from the table. It is written before the window's objects are deleted, so a pass that stops between the two finds the partition already gone and deletes what is left.

func (*Store) PruneCounts

func (s *Store) PruneCounts(ctx context.Context, before time.Time) error

PruneCounts deletes per-minute counts older than before.

func (*Store) RawDeletable

func (s *Store) RawDeletable(ctx context.Context, source string, cutoff time.Time) ([]Window, error)

RawDeletable returns the windows of source whose raw segments may be deleted: compacted at the generation they now hold, so every segment's events are in the Parquet file, and whose last segment was written at or before cutoff. A window owed a compaction is never returned, which is what keeps retention from deleting an event that has not been compacted.

func (*Store) RecordCompacted

func (s *Store) RecordCompacted(ctx context.Context, w Window, c Compaction) error

RecordCompacted records a compaction of w at the generation it was claimed at, and releases the claim. If a segment was recorded since the claim, the window's generation is ahead of the one recorded here and it is claimed again.

func (*Store) RecordCounts

func (s *Store) RecordCounts(ctx context.Context, counts []Count) error

RecordCounts adds counts to the per-minute totals. Counts naming the same source, minute and outcome are summed before they are written, because one upsert cannot touch a row twice. A count for a source that no longer exists is dropped with the rest of that source's data rather than failing the batch.

func (*Store) RecordFailure

func (s *Store) RecordFailure(ctx context.Context, w Window, reason string, hold time.Duration) error

RecordFailure keeps why a compaction of w failed and holds the window back for hold.

func (*Store) RecordLocation

func (s *Store) RecordLocation(ctx context.Context, w Window, resourceID, location string) error

RecordLocation records the resource and partition location of a window before its compaction is recorded, so a failure after the partition moved leaves the record describing where the partition is.

func (*Store) RecordRejections

func (s *Store) RecordRejections(ctx context.Context, rejections []Rejection) error

RecordRejections keeps rejected requests, then trims each source named to its newest maxRejections.

func (*Store) ResourceIDs

func (s *Store) ResourceIDs(ctx context.Context, source string) ([]string, error)

ResourceIDs returns the resources a source's windows were written as.

func (*Store) SourcesForResources

func (s *Store) SourcesForResources(ctx context.Context, ids []string) (map[string]string, error)

SourcesForResources returns, for each resource id that holds a compacted window, the source the window belongs to.

func (*Store) Status

func (s *Store) Status(ctx context.Context, source string, now time.Time) (Status, error)

Status reads a source's status at now.

type Window

type Window struct {
	Source string
	Start  time.Time
	Length time.Duration
	// Generation is the window's segment counter when it was read. Compacting
	// it records this number, so a segment recorded while the compaction ran
	// leaves the window owed another one.
	Generation int64
	// ResourceID and Location are the last compaction's resource and the
	// directory its partition is registered at, empty before the first.
	ResourceID string
	Location   string
	Attempts   int
}

Window is one compaction window of one source.

Jump to

Keyboard shortcuts

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