Documentation
¶
Overview ¶
Package postgres provides command/desired-state transactions and tenant audit history for explicit control-plane composition. NewRuntimeWithPool borrows a go-postgres/v2 pool; NewRuntime preserves the legacy v1 boundary. Apply the hand-maintained embedded MigrationSource separately before use. Constructors do not install/verify schema or own pool shutdown. Driver diagnostics may be wrapped in Go errors; the administrative HTTP layer selects public codes.
Index ¶
- Constants
- Variables
- func MigrationSource() (migrations.Source, error)
- func NewMigrationRunner(database *sql.DB) (*migrations.Runner, error)
- type AuditPage
- type AuditStore
- func (s *AuditStore) AuditSensitiveAccess(ctx context.Context, access controlplane.SensitiveAccess) error
- func (s *AuditStore) ListTenant(ctx context.Context, tenant string, after uint64, limit uint32) (AuditPage, error)
- func (s *AuditStore) RetainBefore(ctx context.Context, tenant string, cutoff time.Time, batchSize uint32) (RetentionResult, error)
- func (s *AuditStore) VerifyTenant(ctx context.Context, tenant string, pageSize uint32) (VerificationReport, error)
- type CommandPage
- type CommandRecord
- type CommandRetentionResult
- type CommandStore
- func (s *CommandStore) Get(ctx context.Context, tenant string, key string) (controlplane.CommandResult, error)
- func (s *CommandStore) ListTenant(ctx context.Context, tenant string, cursor string, limit uint32) (CommandPage, error)
- func (s *CommandStore) RetainCommandsBefore(ctx context.Context, tenant string, cutoff time.Time, batchSize uint32) (CommandRetentionResult, error)
- type DesiredStore
- type Journal
- func (j *Journal) Accept(ctx context.Context, command controlplane.Command) (controlplane.CommandResult, bool, error)
- func (j *Journal) Complete(ctx context.Context, result controlplane.CommandResult) error
- func (j *Journal) MarkAcknowledged(ctx context.Context, result controlplane.CommandResult) error
- func (j *Journal) MarkDispatched(ctx context.Context, result controlplane.CommandResult) error
- type PoolReadiness
- type RetentionResult
- type Runtime
- type VerificationReport
Constants ¶
const ( // MaxCommandPageSize bounds one command-history database response. MaxCommandPageSize uint32 = 1_000 // MaxCommandCursorBytes bounds opaque command-history pagination state. MaxCommandCursorBytes = 1_024 // MaxCommandRetentionBatch bounds one terminal-command cleanup transaction. MaxCommandRetentionBatch uint32 = 1_000 )
const ( // MaxAuditBatch bounds verification and retention database work. MaxAuditBatch uint32 = 1_000 )
Variables ¶
var ( // ErrInvalidAuditRequest reports an unscoped or unbounded audit operation. ErrInvalidAuditRequest = errors.New("postgres: invalid audit request") // ErrAuditAnchorNotFound reports a tenant without initialized audit state. ErrAuditAnchorNotFound = errors.New("postgres: audit anchor not found") // ErrInvalidAuditState reports malformed persisted audit metadata. ErrInvalidAuditState = errors.New("postgres: invalid audit state") )
var ( // ErrInvalidCommandRequest reports an unbounded command lookup scope. ErrInvalidCommandRequest = errors.New("postgres: invalid command request") // ErrInvalidCommandState reports a malformed persisted command result. ErrInvalidCommandState = errors.New("postgres: invalid command state") // ErrInvalidCommandRetentionRequest reports an unsafe cleanup scope. ErrInvalidCommandRetentionRequest = errors.New("postgres: invalid command retention request") // ErrInvalidCommandRetentionState reports an impossible deletion count. ErrInvalidCommandRetentionState = errors.New("postgres: invalid command retention state") )
var ( // ErrNilQueryer reports a missing PostgreSQL read boundary. ErrNilQueryer = errors.New("postgres: queryer is nil") // ErrInvalidDesiredLookup reports an unscoped or unsupported target. ErrInvalidDesiredLookup = errors.New("postgres: invalid desired-state lookup") // ErrDesiredStateNotFound reports a target without durable desired state. ErrDesiredStateNotFound = errors.New("postgres: desired state not found") )
var ( // ErrNilBeginner reports a missing PostgreSQL transaction source. ErrNilBeginner = errors.New("postgres: transaction beginner is nil") // ErrIdempotencyConflict reports reuse of a key for another command. ErrIdempotencyConflict = errors.New("postgres: idempotency key conflicts with stored command") // ErrCompletionConflict reports an attempt to replace a terminal result. ErrCompletionConflict = errors.New("postgres: terminal command result conflicts with stored result") // ErrCommandNotFound reports completion of an unknown tenant-scoped key. ErrCommandNotFound = errors.New("postgres: command not found") // ErrAuditSequenceExhausted reports an audit chain beyond bigint capacity. ErrAuditSequenceExhausted = errors.New("postgres: audit sequence exhausted") // ErrAuditHashInvalid reports malformed persisted audit-chain state. ErrAuditHashInvalid = errors.New("postgres: invalid audit hash") // ErrDesiredStateConflict reports a concurrent desired-state replacement. ErrDesiredStateConflict = errors.New("postgres: desired state update conflict") )
var ErrInvalidMigrationDatabase = errors.New("postgres: migration database is nil")
ErrInvalidMigrationDatabase reports a missing migration connection pool.
var ErrInvalidRuntimePool = errors.New("postgres: invalid runtime pool")
ErrInvalidRuntimePool reports a missing or uninitialized PostgreSQL pool.
var ErrResultNotTerminal = errors.New("postgres: command result is not terminal")
ErrResultNotTerminal rejects accepted results at the completion boundary.
Functions ¶
func MigrationSource ¶
func MigrationSource() (migrations.Source, error)
MigrationSource returns the immutable embedded control-plane schema history.
func NewMigrationRunner ¶
func NewMigrationRunner(database *sql.DB) (*migrations.Runner, error)
NewMigrationRunner builds the bounded migrations runner for the embedded control-plane schema. The caller retains ownership of database.
Types ¶
type AuditPage ¶
type AuditPage struct {
// Entries is a newly materialized sequence-ordered page, verified relative
// to its first stored predecessor.
Entries []history.Entry
// NextSequence is the exclusive cursor for another page; zero means no
// further page was detected.
NextSequence uint64
}
AuditPage is one verified bounded tenant audit-history page.
type AuditStore ¶
type AuditStore struct {
// contains filtered or unexported fields
}
AuditStore verifies and retains tenant audit history transactionally.
func NewAuditStore ¶
func NewAuditStore(beginner gopostgres.Beginner) (*AuditStore, error)
NewAuditStore creates a transaction-backed audit operations repository.
func (*AuditStore) AuditSensitiveAccess ¶
func (s *AuditStore) AuditSensitiveAccess( ctx context.Context, access controlplane.SensitiveAccess, ) error
AuditSensitiveAccess durably records one authorized privileged record read.
func (*AuditStore) ListTenant ¶
func (s *AuditStore) ListTenant( ctx context.Context, tenant string, after uint64, limit uint32, ) (AuditPage, error)
ListTenant returns one internally linked page after the exclusive sequence cursor. It checks contiguity/digests relative to the first stored predecessor, not linkage back to the tenant retention anchor; use VerifyTenant for that traversal. Neither operation authenticates database state against an external trusted root.
func (*AuditStore) RetainBefore ¶
func (s *AuditStore) RetainBefore( ctx context.Context, tenant string, cutoff time.Time, batchSize uint32, ) (RetentionResult, error)
RetainBefore deletes one contiguous old prefix and advances its anchor.
func (*AuditStore) VerifyTenant ¶
func (s *AuditStore) VerifyTenant( ctx context.Context, tenant string, pageSize uint32, ) (VerificationReport, error)
VerifyTenant traverses retained tenant history from its database anchor in a repeatable-read snapshot. pageSize bounds each fetched batch, not total work; callers own the overall context/deadline and external anchor trust.
type CommandPage ¶
type CommandPage struct {
// Records is a newly materialized newest-first page ordered by requested
// time and idempotency key.
Records []CommandRecord
// NextCursor is opaque bounded pagination state; empty means no further
// page was detected.
NextCursor string
}
CommandPage is one bounded newest-first tenant command-history page.
type CommandRecord ¶
type CommandRecord struct {
// Command is the stored envelope, including independently materialized
// option pointers.
Command controlplane.Command
// Result is the stored outcome with materialized optional acknowledgement
// fields.
Result controlplane.CommandResult
}
CommandRecord joins one immutable command envelope to its durable outcome.
type CommandRetentionResult ¶
type CommandRetentionResult struct {
// Deleted counts terminal commands removed after audit/desired-state
// reference refusals.
Deleted uint32
}
CommandRetentionResult reports one bounded terminal-command cleanup batch.
type CommandStore ¶
type CommandStore struct {
// contains filtered or unexported fields
}
CommandStore reads tenant-scoped durable command outcomes.
func NewCommandStore ¶
func NewCommandStore(beginner gopostgres.Beginner) (*CommandStore, error)
NewCommandStore creates a transaction-backed command result reader.
func (*CommandStore) Get ¶
func (s *CommandStore) Get( ctx context.Context, tenant string, key string, ) (controlplane.CommandResult, error)
Get returns one durable result by tenant and idempotency key.
func (*CommandStore) ListTenant ¶
func (s *CommandStore) ListTenant( ctx context.Context, tenant string, cursor string, limit uint32, ) (CommandPage, error)
ListTenant returns one newest-first tenant command-history page.
func (*CommandStore) RetainCommandsBefore ¶
func (s *CommandStore) RetainCommandsBefore( ctx context.Context, tenant string, cutoff time.Time, batchSize uint32, ) (CommandRetentionResult, error)
RetainCommandsBefore deletes old terminal commands that no retained audit event or current desired state references.
type DesiredStore ¶
type DesiredStore struct {
// contains filtered or unexported fields
}
DesiredStore reads tenant-scoped operational state for worker convergence.
func NewDesiredStore ¶
func NewDesiredStore(queryer rowQueryer) (*DesiredStore, error)
NewDesiredStore creates a desired-state read repository.
func (*DesiredStore) Get ¶
func (s *DesiredStore) Get( ctx context.Context, tenant string, target controlplane.Target, ) (control.DesiredRecord, error)
Get returns the latest durable desired state for one target.
type Journal ¶
type Journal struct {
// contains filtered or unexported fields
}
Journal persists command outcomes and their audit records in one transaction.
func NewJournal ¶
func NewJournal(beginner gopostgres.Beginner) (*Journal, error)
NewJournal creates a journal backed by postgres transaction semantics.
func (*Journal) Accept ¶
func (j *Journal) Accept( ctx context.Context, command controlplane.Command, ) (controlplane.CommandResult, bool, error)
Accept atomically records a new command, applicable desired-state revision and initial audit event. Exact duplicates return stored results without a new event or desired revision; changed content conflicts. CommandID alone does not select a different envelope, and persisted times normalize to UTC microseconds.
func (*Journal) Complete ¶
func (j *Journal) Complete( ctx context.Context, result controlplane.CommandResult, ) error
Complete atomically records a terminal command result and audit event. Core transaction work receives at most five seconds, or the earlier deadline of the context passed here; values and cancellation are preserved. Service intentionally detaches pre-dispatch cancellation before calling Complete. PostgreSQL rollback has its separate five-second cleanup budget; cooperative driver I/O means this is not a forced total end-to-end completion limit.
func (*Journal) MarkAcknowledged ¶
func (j *Journal) MarkAcknowledged(ctx context.Context, result controlplane.CommandResult) error
MarkAcknowledged atomically persists a valid data-plane acknowledgement and audit before terminal completion. Acknowledgement is distinct from business success: a reported failed/partial outcome may still have been acknowledged.
func (*Journal) MarkDispatched ¶
func (j *Journal) MarkDispatched(ctx context.Context, result controlplane.CommandResult) error
MarkDispatched atomically persists the pre-dispatch recovery boundary and its audit event.
type PoolReadiness ¶
type PoolReadiness struct {
// contains filtered or unexported fields
}
PoolReadiness adapts postgres bounded health checks to the API contract.
type RetentionResult ¶
type RetentionResult struct {
// Deleted counts entries deleted in this bounded transaction.
Deleted uint32
// AnchorSequence is the last retained-away sequence after prefix deletion.
AnchorSequence uint64
// AnchorHash preserves the digest needed to link later retained history.
AnchorHash history.Hash
// RetainedThrough is the monotonic UTC retention cutoff represented by the
// removed prefix.
RetainedThrough time.Time
}
RetentionResult describes one bounded retained-prefix cleanup batch.
type Runtime ¶
type Runtime struct {
// Journal coordinates command, desired-state and audit writes against the
// borrowed pool.
Journal *Journal
// Audit provides tenant audit reads, verification and retention.
Audit *AuditStore
// Commands provides scoped command lookup, pagination and retention.
Commands *CommandStore
// Desired reads latest durable desired state; it does not drive worker
// convergence.
Desired *DesiredStore
// Readiness pings the borrowed pool through its configured health-check
// budget.
Readiness *PoolReadiness
}
Runtime is the complete PostgreSQL-backed control-plane service bundle. The caller retains ownership of the underlying pool.
func NewRuntime ¶
func NewRuntime(pool *gopostgres.Pool) (*Runtime, error)
NewRuntime wires all control-plane persistence services to a legacy v1 pool. The caller retains ownership; use NewRuntimeWithPool for go-postgres/v2.
func NewRuntimeWithPool ¶
func NewRuntimeWithPool(pool *gopostgresv2.Pool) (*Runtime, error)
NewRuntimeWithPool wires persistence services to a go-postgres/v2 pool. It refuses nil or uninitialized pools. The caller retains pool ownership; transactions use the native pgxpool, and readiness uses the pool's bounded Ping.
type VerificationReport ¶
type VerificationReport struct {
// Events counts retained events traversed, excluding an already retained-away prefix.
Events uint64
// HeadSequence is the last verified retained sequence or anchor sequence
// for an empty chain.
HeadSequence uint64
// HeadHash is the corresponding database-backed chain digest, not external
// authentication.
HeadHash history.Hash
}
VerificationReport identifies the verified tenant audit head.