Documentation
¶
Overview ¶
Package postgres is the default index and the shared deduplication store.
It holds two things that look unrelated: the projection a search reads, which observe writes by cursor (docs/decisions/0020), and the table that lets several writer replicas agree about what has already been written. They share a database and one migration chain, and nothing else: the writer's role may touch the deduplication table and the registry and not the index, observe's the index and not the deduplication table, and the query service's only reads it (GrantWriter, GrantObserver, GrantReader).
Index ¶
- Constants
- Variables
- func AsTenants(ctx context.Context, db DB, pin Pin, fn func(*Index) error) error
- func BindKeyDirectory(ctx context.Context, db DB, id string) error
- func CheckVersion(ctx context.Context, db DB) error
- func GrantObserver(ctx context.Context, db DB, role string) error
- func GrantPurger(ctx context.Context, db DB, role string) error
- func GrantReader(ctx context.Context, db DB, role string) error
- func GrantRoles(ctx context.Context, db DB, r Roles) error
- func GrantWriter(ctx context.Context, db DB, role string) error
- func Migrate(ctx context.Context, db DB) error
- func Schema() string
- type DB
- type Dedupe
- type Index
- func (i *Index) Advance(ctx context.Context, profile, tenant, lastKey string, rows []index.Row) error
- func (i *Index) Capabilities() index.Capabilities
- func (i *Index) Cursor(ctx context.Context, profile, tenant string) (string, error)
- func (i *Index) Facets(ctx context.Context, q index.Query, fields []string, limit int) ([]index.Facet, error)
- func (i *Index) Get(ctx context.Context, profile, id string) (index.Row, index.Provenance, error)
- func (i *Index) Index(ctx context.Context, profile string, rows []index.Row) error
- func (i *Index) Purge(ctx context.Context, profile string, before time.Time, what index.Scope) error
- func (i *Index) ResetCursor(ctx context.Context, profile, tenant string) error
- func (i *Index) Search(ctx context.Context, q index.Query) (index.Page, error)
- type Pin
- type Roles
Constants ¶
const ( // TenantsSetting holds a JSON array of the tenants a connection may see. TenantsSetting = "audit.tenant_ids" // audit:not-an-action — a Postgres setting // AllTenantsSetting, when "on", lets a connection see every tenant. AllTenantsSetting = "audit.all_tenants" // audit:not-an-action — a Postgres setting )
The settings the row-level security policies read. Both are local to a transaction when AsTenants sets them.
const Version = 6
Version is the schema this build expects. A writer whose database is at a different version refuses to start rather than guess: migrating from several replicas at once is a race, so the migration is its own step and this is the check that it ran.
Variables ¶
var ErrKeyDirectory = errors.New("postgres: this writer's key directory is not the deployment's")
ErrKeyDirectory is a writer arriving with a key directory other than the one this deployment's writers share.
Functions ¶
func AsTenants ¶
AsTenants runs a read with the connection pinned, so that the row-level security policies bind it.
The pin is made inside a transaction with the local flag, which Postgres releases when the transaction ends however it ends. The obvious alternative, set_config(name, value, false), is session-scoped, and a pooled connection carries it back into the pool: the next borrower would read under the previous caller's tenants. The transaction is always rolled back, because nothing here writes.
The policies refuse by default: a reader connection that sets nothing sees nothing. This is the second line, under the service's own narrowing; it holds when a query forgets its tenant term, not when the grant itself is wrong.
func BindKeyDirectory ¶
BindKeyDirectory registers a key directory's identity as the deployment's, or checks that it already is.
The first writer to start registers its directory. Every later one, on every start, must bring the same: a writer with its own directory would mint its own keys and hand the same person a second pseudonym, and a directory that was lost and recreated would do the same to every tenant at once. Refusing is the only safe answer, because pseudonyms minted from the wrong keys are in the archive under a lock the moment they are written.
func CheckVersion ¶
CheckVersion reports whether the database holds the schema this build knows.
func GrantObserver ¶ added in v0.7.0
GrantObserver gives a role what the indexer needs: the index and its cursors, and the one function that creates a month's partitions, because the role is not the owner and creating a partition takes the owner. It gets nothing of the deduplication table or the registry.
func GrantPurger ¶ added in v0.7.0
GrantPurger gives a role what the retention job needs: to forget index rows and cursors, and old deduplication entries. It cannot add a row to either.
func GrantReader ¶
GrantReader gives a role what a reading service needs and nothing more: usage on the index's schema, and select on the index's tables, which a month's partition inherits from its parent. Run it as the owner, after Migrate.
The role must exist, and must not be the owner: row-level security does not apply to the role that owns the tables, so tenant isolation for a reader that owned them would rest on the service alone.
func GrantRoles ¶ added in v0.7.0
GrantRoles grants each named role what its part needs and takes from it everything else on this schema's tables. Run it as the owner, after Migrate.
It is the whole of the separation between the parts at the database. The writer, observe and the query service connect as three different roles, none of them the owner, so that a writer that is compromised cannot rewrite what search answers with, an observer cannot mark a record as already written, and a reader can do neither.
func GrantWriter ¶ added in v0.7.0
GrantWriter gives a role what the write path needs: the deduplication table, the catalogue registry and the key directory, and the schema version it checks at start-up. It gives nothing of the index, and takes back anything an earlier version of the deployment had granted: the writer does not index.
Types ¶
type DB ¶
type DB interface {
Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
Begin(ctx context.Context) (pgx.Tx, error)
}
DB is the part of a pgx pool this package uses. Taking an interface keeps the pool's construction — its size, its timeouts, its credentials — where a deployment can see it.
type Dedupe ¶
type Dedupe struct {
DB DB
// Window is how long an identifier is remembered. Default 14 days. It wants
// to be at least as long as the longest redelivery a deployment's stream
// permits, or a repeat arrives after the memory of it has gone.
Window time.Duration
// Now is the clock, for tests.
Now func() time.Time
}
Dedupe is the shared record of what has already been written.
With the deduplication in one process, a repeat is only absorbed by the replica that saw the original, so the writer refuses to start more than one. This table is what lifts the refusal: two replicas consulting it agree about what has been written, and a redelivery landing on either is absorbed.
Asking and marking are separate, and the writer marks only once the copies are durable. See the Dedupe interface in the writer for why that order is the one an audit trail can afford.
func (*Dedupe) Mark ¶
Mark implements the writer's Dedupe.
The timestamp is refreshed on a repeat so that an identifier stays remembered for a window from the last time it was written, not from the first.
type Index ¶
type Index struct {
DB DB
// Partitions is how far ahead a partition is created when one is needed.
// Zero means the month itself only.
Partitions int
// contains filtered or unexported fields
}
Index is the projection.
func NewReader ¶
NewReader returns an index whose every read is pinned to the tenants its query names.
This is what a reading service uses. Over a connection whose role owns the tables the pin changes nothing, since an owner bypasses the policies; over a reader role it is what makes them apply. So it is safe to use either way, and a deployment gets the second line by choosing the role, not by rebuilding the service.
Get takes no tenants — index.Searcher gives it none — so it is pinned to every tenant and the caller checks the row it gets back, as the query service does. A lookup by identifier has no tenant term to forget.
func (*Index) Advance ¶ added in v0.7.0
func (i *Index) Advance(ctx context.Context, profile, tenant, lastKey string, rows []index.Row) error
Advance is what observe indexes with: the rows, and the cursor of the prefix they came from, in one transaction. A crash leaves both or neither, which is what lets a restarted observe resume from the cursor without a row missing behind it. The rows may be empty: a stretch of the listing that holds nothing to index still moves the cursor.
func (*Index) Capabilities ¶
func (i *Index) Capabilities() index.Capabilities
Capabilities implements index.Searcher.
func (*Index) Cursor ¶ added in v0.7.0
Cursor is the key of the last object indexed under a profile and tenant, or "" when none has been.
func (*Index) Facets ¶
func (i *Index) Facets( ctx context.Context, q index.Query, fields []string, limit int, ) ([]index.Facet, error)
Facets counts values under the same filter.
The counts table answers the common case — a whole profile, no filter, which is what a viewer's navigation asks on every page — without touching the events at all. A filtered facet cannot use it, because the table counts rows and not rows-matching-something, so that falls back to counting the events.
func (*Index) Get ¶
Get returns one row and where it came from.
The provenance is the point: an answer a reader can check against the copy the digest chain accounts for is worth more than one they have to believe.
func (*Index) Index ¶
Index implements index.Indexer.
The whole batch is one transaction, and the facet counts move only for the rows the insert actually created. That is why counting is not a call of its own: a caller cannot tell a re-delivered record from a new one, and only the transaction that inserted the row can.
func (*Index) Purge ¶
func (i *Index) Purge(ctx context.Context, profile string, before time.Time, what index.Scope) error
Purge implements index.Indexer.
Identifying forgets who, and keeps what happened; Everything drops the month outright, which is what makes an expired retention cheap. Neither touches the archive: those objects are under a lock, and the profile's own retention is what releases them.
func (*Index) ResetCursor ¶ added in v0.7.0
ResetCursor forgets where a profile's tenant was read to, or every tenant of the profile when tenant is empty, so that observe reads it again from the start. The index keeps what it has: indexing is idempotent, and a row that is there is left alone.
type Pin ¶
Pin is what a connection is allowed to see. It is a grant's tenant half, in the shape the policies read.
type Roles ¶ added in v0.7.0
type Roles struct {
// Writer is the write path's: the deduplication table, the catalogue
// registry and the key directory, and nothing of the index.
Writer string
// Observe is the indexer's: it reads and writes the index and its cursors,
// and creates monthly partitions through one function.
Observe string
// Reader is the query service's: select on the index, bound by row-level
// security to the tenants of each request.
Reader string
// Purge is the retention job's: it forgets index rows and old
// deduplication entries.
Purge string
}
Roles are the database roles of a deployment, one per part. Each is optional: a deployment that runs one part under the owner's own role names none, and gets no separation. Every role must already exist.
Source Files