Documentation
¶
Overview ¶
Package atepg is an ate storage backend built on PostgreSQL.
Each table holds native SQL columns for fields SQL must operate on (primary keys, versions, pagination, update/delete preconditions) plus the complete protobuf message, binary-encoded, in a BYTEA column. TLS is configured entirely through the connection string passed to Connect (standard libpq sslmode/sslrootcert/sslcert/sslkey parameters)
Index ¶
- Variables
- type Persistence
- func (p *Persistence) AcquireLease(ctx context.Context, key string) (*store.Lease, error)
- func (p *Persistence) BindActorToWorker(ctx context.Context, workerName string, assignment *ateapipb.ActorAssignment, ...) error
- func (p *Persistence) Close()
- func (p *Persistence) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*ateapipb.Actor, error)
- func (p *Persistence) CreateActorTemplate(ctx context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error)
- func (p *Persistence) CreateAtespace(ctx context.Context, atespace *ateapipb.Atespace) (*ateapipb.Atespace, error)
- func (p *Persistence) CreateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, ...) (*ateapipb.EgressPolicy, error)
- func (p *Persistence) CreateTag(ctx context.Context, tag *ateapipb.Tag) (*ateapipb.Tag, error)
- func (p *Persistence) CreateWorker(ctx context.Context, worker *ateapipb.Worker) (*ateapipb.Worker, error)
- func (p *Persistence) DeleteActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error)
- func (p *Persistence) DeleteActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error)
- func (p *Persistence) DeleteAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error)
- func (p *Persistence) DeleteEgressPolicy(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.EgressPolicy, error)
- func (p *Persistence) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error)
- func (p *Persistence) DeleteWorker(ctx context.Context, name string, pre store.DeletePreconditions) (*ateapipb.Worker, error)
- func (p *Persistence) FindWorkerHostingActor(ctx context.Context, actorUID string) (string, error)
- func (p *Persistence) GetActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error)
- func (p *Persistence) GetActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error)
- func (p *Persistence) GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error)
- func (p *Persistence) GetEgressPolicy(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.EgressPolicy, error)
- func (p *Persistence) GetTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error)
- func (p *Persistence) GetWorker(ctx context.Context, name string) (*ateapipb.Worker, error)
- func (p *Persistence) GetWorkerAssignment(ctx context.Context, workerName, actorUID string) (*ateapipb.ActorAssignment, error)
- func (p *Persistence) ListActorTemplates(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.ActorTemplate], error)
- func (p *Persistence) ListActors(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Actor], error)
- func (p *Persistence) ListAtespaces(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Atespace], error)
- func (p *Persistence) ListTags(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Tag], error)
- func (p *Persistence) ListWorkerAssignments(ctx context.Context, workerName string, opts store.ListOptions) (store.ListResponse[*ateapipb.ActorAssignment], error)
- func (p *Persistence) ListWorkers(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Worker], error)
- func (p *Persistence) ReleaseActorFromWorker(ctx context.Context, workerName string, actorUID string) (*ateapipb.Worker, error)
- func (p *Persistence) UpdateActor(ctx context.Context, actorRef resources.ActorRef, ...) (*ateapipb.Actor, error)
- func (p *Persistence) UpdateActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef, ...) (*ateapipb.ActorTemplate, error)
- func (p *Persistence) UpdateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, ...) (*ateapipb.EgressPolicy, error)
- func (p *Persistence) UpdateTag(ctx context.Context, tagRef resources.TagRef, precondition store.Precondition, ...) (*ateapipb.Tag, error)
- func (p *Persistence) UpdateWorker(ctx context.Context, name string, precondition store.Precondition, ...) (*ateapipb.Worker, error)
- func (p *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, error)
Constants ¶
This section is empty.
Variables ¶
ErrUnavailable reports that ateapi could not establish the initial PostgreSQL connection. Callers can retry this error before startup.
Functions ¶
This section is empty.
Types ¶
type Persistence ¶
type Persistence struct {
// contains filtered or unexported fields
}
func Connect ¶
func Connect(ctx context.Context, dsn, schema string) (*Persistence, error)
Connect opens a pgxpool against dsn, creates schema if necessary, and applies pending schema migrations. A dedicated watch pool isolates outbox polling and maintenance from writes.
func NewPersistence ¶
NewPersistence wraps an already-open pool, applying pending migrations. Callers that already hold a pool (e.g. tests using testcontainers) use this directly instead of Connect; outbox watch traffic shares the given pool.
func (*Persistence) AcquireLease ¶
func (*Persistence) BindActorToWorker ¶
func (p *Persistence) BindActorToWorker(ctx context.Context, workerName string, assignment *ateapipb.ActorAssignment, admit func(*ateapipb.Worker) error) error
func (*Persistence) Close ¶
func (p *Persistence) Close()
Close stops the outbox maintenance loop and waits for it to exit, then closes the watch pool if Connect created one. It does not close the main pool, which the caller owns.
func (*Persistence) CreateActor ¶
func (*Persistence) CreateActorTemplate ¶
func (p *Persistence) CreateActorTemplate(ctx context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error)
func (*Persistence) CreateAtespace ¶
func (*Persistence) CreateEgressPolicy ¶
func (p *Persistence) CreateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, policy *ateapipb.EgressPolicy) (*ateapipb.EgressPolicy, error)
func (*Persistence) CreateWorker ¶
func (*Persistence) DeleteActor ¶
func (*Persistence) DeleteActorTemplate ¶
func (p *Persistence) DeleteActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error)
func (*Persistence) DeleteAtespace ¶
func (*Persistence) DeleteEgressPolicy ¶
func (p *Persistence) DeleteEgressPolicy(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.EgressPolicy, error)
func (*Persistence) DeleteWorker ¶
func (p *Persistence) DeleteWorker(ctx context.Context, name string, pre store.DeletePreconditions) (*ateapipb.Worker, error)
func (*Persistence) FindWorkerHostingActor ¶
func (*Persistence) GetActorTemplate ¶
func (p *Persistence) GetActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error)
func (*Persistence) GetAtespace ¶
func (*Persistence) GetEgressPolicy ¶
func (p *Persistence) GetEgressPolicy(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.EgressPolicy, error)
func (*Persistence) GetWorkerAssignment ¶
func (p *Persistence) GetWorkerAssignment(ctx context.Context, workerName, actorUID string) (*ateapipb.ActorAssignment, error)
func (*Persistence) ListActorTemplates ¶
func (p *Persistence) ListActorTemplates(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.ActorTemplate], error)
func (*Persistence) ListActors ¶
func (p *Persistence) ListActors(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Actor], error)
func (*Persistence) ListAtespaces ¶
func (p *Persistence) ListAtespaces(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Atespace], error)
func (*Persistence) ListTags ¶
func (p *Persistence) ListTags(ctx context.Context, atespace string, opts store.ListOptions) (store.ListResponse[*ateapipb.Tag], error)
func (*Persistence) ListWorkerAssignments ¶
func (p *Persistence) ListWorkerAssignments(ctx context.Context, workerName string, opts store.ListOptions) (store.ListResponse[*ateapipb.ActorAssignment], error)
func (*Persistence) ListWorkers ¶
func (p *Persistence) ListWorkers(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Worker], error)
func (*Persistence) ReleaseActorFromWorker ¶
func (*Persistence) UpdateActor ¶
func (*Persistence) UpdateActorTemplate ¶
func (p *Persistence) UpdateActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef, precondition store.Precondition, mutate func(*ateapipb.ActorTemplate) error) (*ateapipb.ActorTemplate, error)
func (*Persistence) UpdateEgressPolicy ¶
func (p *Persistence) UpdateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, precondition store.Precondition, mutate func(*ateapipb.EgressPolicy) error) (*ateapipb.EgressPolicy, error)
func (*Persistence) UpdateWorker ¶
func (p *Persistence) UpdateWorker(ctx context.Context, name string, precondition store.Precondition, mutate func(*ateapipb.Worker) error) (*ateapipb.Worker, error)
UpdateWorker runs mutate against the worker read FOR UPDATE inside the write transaction, so a concurrent writer blocks on the row lock rather than interleaving. That is what makes an occupancy test inside mutate a compare-and-set. The predicate cannot be pushed into SQL: the row stores an opaque marshaled proto, so assignment is not addressable in a WHERE clause.
func (*Persistence) WatchWorkers ¶
func (p *Persistence) WatchWorkers(ctx context.Context) (*store.WorkerWatch, error)
WatchWorkers subscribes by polling the worker_outbox table using an xid cursor. It fences reads behind pg_snapshot_xmin (the oldest in-flight transaction) to guarantee gap-free delivery. Note that a long-running transaction anywhere in the database will stall delivery.
Events are delivered in xid order, so consumers must reconcile worker versions. If the watcher detects missed events—either by lagging behind retention drops or if a database restart truncates the UNLOGGED partitions—it closes the channel to force the consumer to resync from the primary tables.
Source Files
Click to show internal directories.
Click to hide internal directories.