atepg

package
v0.5.0 Latest Latest
Warning

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

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

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

Constants

This section is empty.

Variables

View Source 
var ErrUnavailable = errors.New("PostgreSQL is unavailable")

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 ConnectConfig added in v0.4.0 

type ConnectConfig struct {
	ReadWriteDSN  string
	OwnerDSN      string
	ReadWriteRole string
	OwnerRole     string
	Schema        string
	PoolMaxConns  int32
}

ConnectConfig configures the PostgreSQL pools used by Persistence.

type Persistence 

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

func Connect 

func Connect(ctx context.Context, config ConnectConfig) (*Persistence, error)

Connect opens the read/write, owner, watch, and probe pools. It creates the schema and applies migrations.

func NewPersistence 

func NewPersistence(ctx context.Context, pool *pgxpool.Pool) (*Persistence, error)

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 and Ping share the given pool.

func (*Persistence) AcquireLease 

func (p *Persistence) AcquireLease(ctx context.Context, key string) (*store.Lease, error)

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 maintenance loop and waits for it to exit, then closes the auxiliary pools if Connect created them. It does not close the main pool, which the caller owns.

func (*Persistence) CreateActor 

func (p *Persistence) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*ateapipb.Actor, error)

func (*Persistence) CreateActorTemplate 

func (p *Persistence) CreateActorTemplate(ctx context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error)

func (*Persistence) CreateAtespace 

func (p *Persistence) CreateAtespace(ctx context.Context, atespace *ateapipb.Atespace) (*ateapipb.Atespace, error)

func (*Persistence) CreateAtespaceAccessPolicy added in v0.4.0 

func (p *Persistence) CreateAtespaceAccessPolicy(ctx context.Context, name string, policy *ateapipb.AccessPolicy) (*ateapipb.AccessPolicy, error)

func (*Persistence) CreateEgressPolicy 

func (p *Persistence) CreateEgressPolicy(ctx context.Context, actorRef resources.ActorRef, policy *ateapipb.EgressPolicy) (*ateapipb.EgressPolicy, error)

func (*Persistence) CreateGlobalAccessPolicy added in v0.4.0 

func (p *Persistence) CreateGlobalAccessPolicy(ctx context.Context, policy *ateapipb.AccessPolicy) (*ateapipb.AccessPolicy, error)

func (*Persistence) CreateTag 

func (p *Persistence) CreateTag(ctx context.Context, tag *ateapipb.Tag) (*ateapipb.Tag, error)

func (*Persistence) CreateWorker 

func (p *Persistence) CreateWorker(ctx context.Context, worker *ateapipb.Worker) (*ateapipb.Worker, error)

func (*Persistence) DeleteActor 

func (p *Persistence) DeleteActor(ctx context.Context, actorRef resources.ActorRef, precondition store.DeletePreconditions) (*ateapipb.Actor, error)

func (*Persistence) DeleteActorTemplate 

func (p *Persistence) DeleteActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef, precondition store.DeletePreconditions) (*ateapipb.ActorTemplate, error)

func (*Persistence) DeleteAtespace 

func (p *Persistence) DeleteAtespace(ctx context.Context, name string, precondition store.DeletePreconditions) (*ateapipb.Atespace, error)

func (*Persistence) DeleteAtespaceAccessPolicy added in v0.4.0 

func (p *Persistence) DeleteAtespaceAccessPolicy(ctx context.Context, name string, precondition store.DeletePreconditions) (*ateapipb.AccessPolicy, error)

func (*Persistence) DeleteEgressPolicy 

func (p *Persistence) DeleteEgressPolicy(ctx context.Context, actorRef resources.ActorRef, precondition store.DeletePreconditions) (*ateapipb.EgressPolicy, error)

func (*Persistence) DeleteTag 

func (p *Persistence) DeleteTag(ctx context.Context, tagRef resources.TagRef, precondition store.DeletePreconditions) (*ateapipb.Tag, error)

func (*Persistence) DeleteWorker 

func (p *Persistence) DeleteWorker(ctx context.Context, name string, precondition store.DeletePreconditions) (*ateapipb.Worker, error)

func (*Persistence) FindWorkerHostingActor 

func (p *Persistence) FindWorkerHostingActor(ctx context.Context, actorUID string) (string, error)

func (*Persistence) GetActor 

func (p *Persistence) GetActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, error)

func (*Persistence) GetActorTemplate 

func (p *Persistence) GetActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error)

func (*Persistence) GetAtespace 

func (p *Persistence) GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error)

func (*Persistence) GetAtespaceAccessPolicy added in v0.4.0 

func (p *Persistence) GetAtespaceAccessPolicy(ctx context.Context, name string) (*ateapipb.AccessPolicy, error)

func (*Persistence) GetEgressPolicy 

func (p *Persistence) GetEgressPolicy(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.EgressPolicy, error)

func (*Persistence) GetGlobalAccessPolicy added in v0.4.0 

func (p *Persistence) GetGlobalAccessPolicy(ctx context.Context) (*ateapipb.AccessPolicy, error)

func (*Persistence) GetTag 

func (p *Persistence) GetTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error)

func (*Persistence) GetWorker 

func (p *Persistence) GetWorker(ctx context.Context, name string) (*ateapipb.Worker, 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) Ping added in v0.5.0 

func (p *Persistence) Ping(ctx context.Context) error

Ping reports whether PostgreSQL answers. It uses a connection reserved for it, so a saturated request pool cannot make it fail: it fails only when the database does not answer within ctx.

func (*Persistence) Pool added in v0.2.0 

func (p *Persistence) Pool() *pgxpool.Pool

Pool returns the underlying PostgreSQL connection pool.

func (*Persistence) ReleaseActorFromWorker 

func (p *Persistence) ReleaseActorFromWorker(ctx context.Context, workerName string, actorUID string) (*ateapipb.Worker, error)

func (*Persistence) SetPolicyManager added in v0.3.0 

func (p *Persistence) SetPolicyManager(pm *authz.PolicyManager)

SetPolicyManager configures the authorization policy manager that writes OpenFGA tuples in the same transaction as access policy and atespace mutations. It must be set before the store serves access policy or atespace writes.

func (*Persistence) UpdateActor 

func (p *Persistence) UpdateActor(ctx context.Context, actorRef resources.ActorRef, precondition store.Precondition, mutate func(*ateapipb.Actor) error) (*ateapipb.Actor, error)

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) UpdateAtespaceAccessPolicy added in v0.4.0 

func (p *Persistence) UpdateAtespaceAccessPolicy(ctx context.Context, name string, precondition store.Precondition, mutate func(*ateapipb.AccessPolicy) error) (*ateapipb.AccessPolicy, 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) UpdateGlobalAccessPolicy added in v0.4.0 

func (p *Persistence) UpdateGlobalAccessPolicy(ctx context.Context, precondition store.Precondition, mutate func(*ateapipb.AccessPolicy) error) (*ateapipb.AccessPolicy, error)

func (*Persistence) UpdateTag 

func (p *Persistence) UpdateTag(ctx context.Context, tagRef resources.TagRef, precondition store.Precondition, mutate func(*ateapipb.Tag) error) (*ateapipb.Tag, 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.

This process's own writes are published at commit, out of xid order, and again on the poll, so consumers must reconcile versions and tolerate duplicates.

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

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.