Documentation
¶
Index ¶
- Constants
- Variables
- func AppendKey(dst, prefix []byte, k Key) []byte
- func IdempotencyKey(id tenant.ID, k Key) string
- func KeyPrefix(id tenant.ID) []byte
- type Claim
- type Deduplicator
- type Dynamo
- type DynamoConfig
- type Embedded
- type Factory
- type Key
- type Managed
- func (m *Managed) Apply(enabled bool) error
- func (m *Managed) Close() error
- func (m *Managed) Commit(ctx context.Context, claims []Claim, retention time.Duration) error
- func (m *Managed) Open() bool
- func (m *Managed) Release(ctx context.Context, claims []Claim) error
- func (m *Managed) Reserve(ctx context.Context, keys []Key, lease time.Duration) ([]Claim, error)
- type Status
- type Stores
Constants ¶
const DefaultLease = 30 * time.Second
DefaultLease is how long a Claimed key stays pending when the caller names no lease: long enough to cover a publish, short enough that a request that died mid-publish does not hold the id for long.
const ( // MaxIDBytes is the longest escaped id stored verbatim: a DynamoDB // partition key holds at most 2,048 bytes, and the tenant and table share // them. MaxIDBytes = 1024 )
The key every backend stores is text:
<tenant>/<table>/<id> acme/clicks/evt-123 <tenant>/<table>/#<sha256> an id too long to store verbatim
The table and id are escaped and joined by internal/keyenc, which never writes '/' or '#', and a tenant id holds neither (tenant.Parse), so the fields split back apart, a table name may hold any byte, and no two (tenant, table, id) triples share a key. A key is ASCII, so it is a valid DynamoDB String, and holds no NUL, so it never meets a tenant ‖ NUL ‖ id key written before #222.
Variables ¶
var ErrCreateTableNeedsEndpoint = errors.New("dedupe: create_table is for dynamodb-local only; set the endpoint")
ErrCreateTableNeedsEndpoint refuses CreateTable against real AWS: the production table belongs to the deployment's infrastructure code.
var ErrDisabled = errors.New("dedupe is disabled")
ErrDisabled is returned by Managed's calls while dedupe is switched off. The ingest handler consults the settings snapshot before calling, so it only sees this in the window of a reload that flips dedupe.enabled: the snapshot and the store transition at different instants, and a record caught between them is published un-deduped rather than failed.
ErrUnavailable is returned by Managed's calls when dedupe is switched on but the store failed to open, and wrapped by a backend's error when a retry later can succeed. Ingest fails closed on it — the settings asked for dedupe, so publishing un-deduped is not a fallback.
Functions ¶
func AppendKey ¶ added in v0.2.0
AppendKey appends k's stored form, under the tenant prefix from KeyPrefix, to dst.
func IdempotencyKey ¶ added in v0.2.0
IdempotencyKey is k's message id for the queue under tenant id: the first 128 bits of the stored key's SHA-256, in hex, so a republished record is recognised without its id riding in a header verbatim.
Types ¶
type Claim ¶ added in v0.2.0
Claim is Reserve's answer for one key. Token is the backend's proof of ownership, opaque to callers; Release compares it.
type Deduplicator ¶
type Deduplicator interface {
Reserve(ctx context.Context, keys []Key, lease time.Duration) ([]Claim, error)
Commit(ctx context.Context, claims []Claim, retention time.Duration) error
Release(ctx context.Context, claims []Claim) error
Close() error
}
Deduplicator is a tenant's store of seen ids. Callers reach every backend through Managed, which hands a backend distinct keys, a lease > 0, and only Claimed claims to Commit and Release — a backend may assume all three, and Managed's callers get the behaviour below either way.
Reserve is atomic per key: of any number of concurrent Reserves for the same key — in this process or any other sharing the backend — at most one returns Claimed. It returns one Claim per key, in input order. On error it has released every claim it knows it made; a write whose outcome the error left unknown (a timeout, a cancelled call) may still land afterwards, and then holds its key InFlight until the lease ends, like an abandoned claim. The error wraps ErrUnavailable when retrying later can succeed (throttled, timed out, backend unreachable).
Commit makes Claimed claims duplicates for retention (0 = no expiry) and ignores claims of any other status. It is unconditional: a commit that lands after its lease lapsed and another request re-claimed the key is still correct, because the committing request did publish.
Release gives up the Claimed claims it still owns (token match); a claim that has lapsed or been re-claimed is left alone.
There is deliberately no read-only check: every caller that asks "have I seen this" needs the claim too, and a separate read is how #390 happened.
type Dynamo ¶ added in v0.2.0
type Dynamo struct {
// contains filtered or unexported fields
}
Dynamo is the DynamoDB implementation: every tenant's keys in one shared table, so pods sharing the table share seen ids and Reserve's conditional write is atomic across all of them. WaveHouse never creates the table in production; CreateTable is for dynamodb-local.
func NewDynamo ¶ added in v0.2.0
func NewDynamo(ctx context.Context, cfg DynamoConfig, extra ...func(*config.LoadOptions) error) (*Dynamo, error)
NewDynamo builds the backend over a client from the SDK's default config chain. extra is appended to the chain's options (a test's static credentials or HTTP client, say). It dials nothing: Check does.
func (*Dynamo) Check ¶ added in v0.2.0
Check verifies the table exists with the key schema this backend writes: pk, a string, as the only key. TTL not enabled on ex is logged, not refused: expiry never depends on it, only storage does.
func (*Dynamo) CreateTable ¶ added in v0.2.0
CreateTable creates the table on dynamodb-local, with TTL on ex, and waits for it. A table that already exists is left as it is. Its errors are classified as every call's are, so an endpoint not up yet is ErrUnavailable.
type DynamoConfig ¶ added in v0.2.0
type DynamoConfig struct {
// Table is the shared table every tenant's keys live in. Required.
Table string
// Region overrides the SDK chain's region (AWS_REGION) when set.
Region string
// Endpoint points the client at dynamodb-local. Tests and development
// only; it is also what unlocks CreateTable.
Endpoint string
// Timeout bounds each DynamoDB call, its SDK retries included.
// 0 = 250ms. The retries' jittered backoff is capped so that together
// it waits at most half of Timeout (retryBackoff).
Timeout time.Duration
// MaxAttempts is the SDK retryer's attempts per call. 0 = 3.
MaxAttempts int
// RetryMode is "standard" (default) or "adaptive", which also rate-limits
// the client after throttles.
RetryMode string
// ReserveConcurrency bounds the parallel calls one Reserve, Commit or
// Release makes, and sizes the client's idle connection pool to match
// (never below the SDK's default of 10 per host).
// 0 = 64.
ReserveConcurrency int
}
DynamoConfig is the DynamoDB backend's wiring. Credentials are never here: the SDK's default chain finds them (EKS Pod Identity or IRSA in a pod, the environment or a profile locally).
type Embedded ¶ added in v0.2.0
type Embedded struct {
// contains filtered or unexported fields
}
Embedded is the embedded implementation: every tenant's seen ids in one Pebble instance at data_dir/pebble, each key led by its tenant (#583 story 3) and then its table (#222), with the pending claims in memory beside it (pendingSet). One instance means a thousand tenants cost one instance's goroutines, open files and heap rather than a thousand. The instance opens with the first tenant's store switched on and closes with the last one switched off: it is open exactly while some tenant has dedupe on, and a tenant switched off, rejected or removed keeps its seen ids for when it is back. Pebble is one process's, so two pods on it do not share seen ids.
func NewEmbedded ¶
NewEmbedded returns the embedded implementation under dataDir. Nothing is opened until a tenant's store is.
func (*Embedded) Open ¶ added in v0.2.0
Open reports whether the instance is open: whether some tenant's store is.
type Factory ¶ added in v0.2.0
Factory builds one tenant's store, closed: Stores calls it the first time a tenant is named. Whether tenants share a backend is the factory's business — Embedded.Tenant keeps them all in one Pebble instance — so nothing that holds the Stores changes with it.
func (Factory) Gated ¶ added in v0.2.0
Gated returns a Factory whose stores open only once ready returns nil, its error being the open's: a store switched on meanwhile stays closed and fails closed (ErrUnavailable) until an Apply finds the backend ready. For a backend whose tenant opens are free but whose shared resource (a remote table) is checked once.
type Key ¶ added in v0.2.0
Key is one record's dedupe identity inside a tenant's store. The tenant is bound by the store (Stores.For), so a Key never carries it.
type Managed ¶ added in v0.2.0
type Managed struct {
// contains filtered or unexported fields
}
Managed is a Deduplicator whose backing store follows the hot-reloadable dedupe.enabled setting: Apply(true) opens it through the function NewManaged was given, Apply(false) closes it, and in-flight Reserve, Commit and Release calls are serialized against that swap so a reload can never close the store under a lookup. Which store that is — a tenant's share of the embedded Pebble instance (Embedded.Tenant), a remote backend's view later — is the opener's business, so every backend gets the same switch semantics.
func NewManaged ¶ added in v0.2.0
func NewManaged(open func() (Deduplicator, error)) *Managed
NewManaged returns a closed Managed store over open. Nothing is opened until Apply(true).
func (*Managed) Apply ¶ added in v0.2.0
Apply reconciles the store with the desired state, idempotently: an already-open store stays open, an already-closed one stays closed. A failed open leaves the store closed and returns the error — the caller decides whether that is fatal (boot) or a logged degradation (reload).
A no-op call — the desired state already holds — returns under the read lock alone; only a real transition takes the write lock, re-checked once held in case another Apply won the race. This matters because a settings reload calls Apply for every tenant under the registry lock: on a network backend, Commit and Release can hold the read lock for as long as an outage lasts, and the write lock waits out every reader, so an unconditional write lock here would serialize the whole reload behind them, tenant after tenant.
func (*Managed) Close ¶ added in v0.2.0
Close releases the store if open. Safe to call when already closed.
func (*Managed) Commit ¶ added in v0.2.0
Commit delegates the Claimed claims to the open store, with Reserve's switch semantics.
func (*Managed) Release ¶ added in v0.2.0
Release delegates the Claimed claims to the open store, with Reserve's switch semantics.
func (*Managed) Reserve ¶ added in v0.2.0
Reserve collapses a key repeated inside keys to one backend claim — later occurrences answer Duplicate — reads a lease <= 0 as DefaultLease, and delegates the rest to the open store; ErrDisabled while switched off, ErrUnavailable while switched on but not open.
type Status ¶ added in v0.2.0
type Status uint8
Status is Reserve's verdict for one key.
const ( // Claimed is a first sighting within retention. The caller now holds a // pending claim and must Commit it once the record is published, or // Release it if the publish definitely failed. An abandoned claim lapses // after the lease. Claimed Status = iota + 1 // Duplicate means the key was committed earlier and has not expired: skip // the record. Managed also answers it for a key repeated inside one // Reserve call, after its first occurrence, whatever the first answered. Duplicate // InFlight means another request holds a live claim on the key. Its // outcome is not known yet, so the caller answers 503 and the client // retries. InFlight )
type Stores ¶ added in v0.2.0
type Stores struct {
// contains filtered or unexported fields
}
Stores is one Managed store per tenant (#583 story 7), each following its own tenant's dedupe.enabled through Apply. A store is built on first use and forgotten by Retain once its tenant is no longer served; its seen ids stay either way, so a tenant whose folder comes back finds them where it left them.
func NewStores ¶ added in v0.2.0
NewStores returns an empty Stores that builds each tenant's store with build.
func (*Stores) Close ¶ added in v0.2.0
Close closes every store, in id order, and reports the failures joined. Safe to call more than once.
func (*Stores) For ¶ added in v0.2.0
For returns id's store, building it closed on first use. So a tenant a reload adopted a moment ago has a store before the hook that follows its switch runs — one answering ErrDisabled, the reload-window semantics Managed already has — rather than none.
func (*Stores) Retain ¶ added in v0.2.0
Retain closes and forgets every store whose tenant keep does not name — a tenant the registry no longer serves — and touches nothing on disk. The map is edited under the lock and the stores closed outside it, so one tenant's close (the last one closes the Pebble instance, which waits on its flushes and compactions) never stalls another tenant's lookup. The close failures are joined; the stores are forgotten either way. A store For builds for a dropped tenant meanwhile is closed and stays so — only the reconcile that called Retain opens one — so a tenant never has two stores open at once.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package dedupetest is the conformance suite every dedupe backend runs: the Deduplicator contract (dedupe.go) as tests, driven through the production path — a backend's Factory and the Managed switch it returns — so a backend that passes here behaves the same under ingest as every other.
|
Package dedupetest is the conformance suite every dedupe backend runs: the Deduplicator contract (dedupe.go) as tests, driven through the production path — a backend's Factory and the Managed switch it returns — so a backend that passes here behaves the same under ingest as every other. |