postgres

package
v0.1.46 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func SessionAppendLockKey added in v0.1.10 

func SessionAppendLockKey(sessionID string) int64

SessionAppendLockKey derives the Postgres advisory lock key that serializes every write to one Session's durable event stream. All append paths in this package take pg_advisory_xact_lock on this key before touching session_events or session_event_streams.

This is exported because the key is a cross-module contract, not an implementation detail. session_events and session_event_streams carry no foreign key to agent_sessions, so a row-level FOR UPDATE on the Session row does NOT conflict with an append: an FK insert would take FOR KEY SHARE on the parent row, and with no FK there is nothing to take. Any component that removes a Session's durable events must therefore take this same lock, in the same transaction, before deleting them. Otherwise a concurrent Append commits an event row that outlives its Session as a permanently unreachable orphan still carrying its payload.

Lock order. Every path in this package takes this lock BEFORE any row lock, session_adapter_connections included. Callers MUST do the same. There is one global order and no exception to it: 

this advisory lock -> session_adapter_connections row lock -> every other row lock

The order matters because three paths here take this lock and then write session_adapter_connections: appendEventsInTx via FenceAttentionTerminal on a terminal event (reachable from the exported Append), TerminateAdapterConnectionBeforeHello, and acceptAdapterHelloWithWriterLeaseTx. A caller that row-locks session_adapter_connections before taking this lock closes a cycle against all three, and Postgres aborts one side with SQLSTATE 40P01 -- failing either the caller or a durable event append.

AppendAdapterEvents formerly inverted the order, row-locking session_adapter_connections first so adapter authority was revalidated before any event-stream work. It now takes this lock first and revalidates second. That preserves the guarantee that no event is appended without valid authority -- the row is still checked before the append, and rechecked after it in the same transaction -- while removing the only ordering exception. TestAppendAdapterEventsLocksEventStreamBeforeAuthority pins the current order and fails with SQLSTATE 55P03 if the old one is restored.

Types

type AttentionBackfillCheckpoint added in v0.1.7 

type AttentionBackfillCheckpoint struct{ AfterSessionID string }

type AttentionBackfillCheckpointStore added in v0.1.7 

type AttentionBackfillCheckpointStore interface {
	Load(context.Context) (AttentionBackfillCheckpoint, error)
	Save(context.Context, AttentionBackfillCheckpoint) error
}

type AttentionBackfillResult added in v0.1.7 

type AttentionBackfillResult struct {
	Checkpoint AttentionBackfillCheckpoint
	Processed  int
	Incomplete int
	Done       bool
}

type FileAttentionBackfillCheckpointStore added in v0.1.7 

type FileAttentionBackfillCheckpointStore struct{ Path string }

func (FileAttentionBackfillCheckpointStore) Load added in v0.1.7 

func (s FileAttentionBackfillCheckpointStore) Load(context.Context) (AttentionBackfillCheckpoint, error)

func (FileAttentionBackfillCheckpointStore) Save added in v0.1.7 

func (s FileAttentionBackfillCheckpointStore) Save(_ context.Context, c AttentionBackfillCheckpoint) error

type Store 

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

func New 

func New(pool *pgxpool.Pool) *Store

func NewAdapterConnectionTx added in v0.1.7 

func NewAdapterConnectionTx(tx pgx.Tx) *Store

NewAdapterConnectionTx binds connection operations to a caller-owned transaction. The caller alone commits or rolls it back.

func NewEventStoreTx added in v0.1.10 

func NewEventStoreTx(tx pgx.Tx) *Store

NewEventStoreTx binds EventStore mutations to a caller-owned transaction. The caller alone commits or rolls back the transaction. This is used when a platform-owned lifecycle transition must commit its durable Session event and companion rows atomically.

func (*Store) AcceptAdapterHello added in v0.1.7 

func (s *Store) AcceptAdapterHello(ctx context.Context, sessionID string, hello store.AdapterHello) (store.AdapterConnection, error)

func (*Store) AcknowledgeFileReferenceDelivery added in v0.1.7 

func (s *Store) AcknowledgeFileReferenceDelivery(ctx context.Context, sessionID, commandID string, version int64, writer store.FileReferenceWriter) (store.FileReferenceCommand, error)

func (*Store) AcknowledgeSettingsCommandDelivery added in v0.1.7 

func (s *Store) AcknowledgeSettingsCommandDelivery(ctx context.Context, sessionID, commandID string, reservationVersion int64, writer store.SettingsWriter) (store.SettingsCommand, error)

func (*Store) ActivateAdapterCredential added in v0.1.7 

func (s *Store) ActivateAdapterCredential(ctx context.Context, sessionID string, activation store.AdapterCredentialActivation) (store.AdapterConnection, error)

func (*Store) AdapterConnection added in v0.1.7 

func (s *Store) AdapterConnection(ctx context.Context, sessionID string) (store.AdapterConnection, error)

func (*Store) AdapterSessionAdmissionTruth added in v0.1.7 

func (s *Store) AdapterSessionAdmissionTruth(ctx context.Context, sessionID string) (store.SessionAdmissionTruth, error)

func (*Store) AllocateAdapterGrantFence added in v0.1.7 

func (s *Store) AllocateAdapterGrantFence(ctx context.Context) (int64, error)

func (*Store) Append 

func (s *Store) Append(ctx context.Context, sessionID string, evs []store.PendingEvent) (firstSeq int64, err error)

func (*Store) AppendAdapterEvents added in v0.1.7 

func (s *Store) AppendAdapterEvents(ctx context.Context, sessionID string, admission store.AdapterConnectionAdmission, events []store.PendingEvent) (int64, error)

func (*Store) AttachAttempt added in v0.1.7 

func (s *Store) AttachAttempt(ctx context.Context, jtiHash [32]byte) (store.AttachAttempt, error)

func (*Store) Attachment added in v0.1.7 

func (s *Store) Attachment(ctx context.Context, attachID string) (store.Attachment, error)

func (*Store) AttachmentForTarget added in v0.1.7 

func (s *Store) AttachmentForTarget(ctx context.Context, targetSessionID string) (store.Attachment, error)

func (*Store) AttentionSnapshot added in v0.1.7 

func (s *Store) AttentionSnapshot(ctx context.Context, sessionIDs []string) ([]store.SessionAttentionSummary, error)

func (*Store) AttentionSummaryPage added in v0.1.7 

func (s *Store) AttentionSummaryPage(ctx context.Context, request store.AttentionSummaryPageRequest) (store.AttentionSummaryPage, error)

func (*Store) BackfillAttentionBatch added in v0.1.7 

func (s *Store) BackfillAttentionBatch(ctx context.Context, checkpoint AttentionBackfillCheckpoint, batchSize int) (AttentionBackfillResult, error)

func (*Store) ClaimPendingCommand added in v0.1.7 

func (s *Store) ClaimPendingCommand(ctx context.Context, sessionID string, authority store.CommandAuthority, commandID string) (store.PendingCommandClaim, error)

func (*Store) CommitAttachAttempt added in v0.1.7 

func (s *Store) CommitAttachAttempt(ctx context.Context, request store.AttachAttemptRequest) (store.AttachAttemptCommit, error)

func (*Store) CommitFileReferenceCommand added in v0.1.7 

func (s *Store) CommitFileReferenceCommand(ctx context.Context, sessionID string, message store.PendingEvent, request store.FileReferenceCommandRequest) (store.FileReferenceCommandReserve, error)

func (*Store) CommitPendingCommand added in v0.1.7 

func (s *Store) CommitPendingCommand(ctx context.Context, sessionID string, authority store.CommandAuthority, event store.PendingEvent, request store.PendingCommandRequest) (store.PendingCommandCommit, error)

func (*Store) CommitProposedEvent added in v0.1.7 

func (s *Store) CommitProposedEvent(ctx context.Context, sessionID string, authority store.CommandAuthority, proposal store.ProposedEventRequest) (store.ProposedEventReceipt, error)

func (*Store) CommitWarmAttach added in v0.1.7 

func (s *Store) CommitWarmAttach(ctx context.Context, request store.WarmAttachRequest) (store.WarmAttachCommit, error)

func (*Store) CreateAttachment added in v0.1.7 

func (s *Store) CreateAttachment(ctx context.Context, request store.AttachmentCreate) (store.AttachmentCommit, error)

func (*Store) ExpireWarmAttach added in v0.1.7 

func (s *Store) ExpireWarmAttach(ctx context.Context, attachID string, expectedDeliveryVersion int64) (store.WarmAttachExpiry, error)

func (*Store) FileReferenceCommand added in v0.1.7 

func (s *Store) FileReferenceCommand(ctx context.Context, sessionID, commandID string) (store.FileReferenceCommand, error)

func (*Store) FinalizeFileReferenceCommand added in v0.1.7 

func (s *Store) FinalizeFileReferenceCommand(ctx context.Context, sessionID, commandID string, finalize store.FileReferenceCommandFinalize) (store.FileReferenceCommand, error)

func (*Store) FinalizeSettingsCommand added in v0.1.7 

func (s *Store) FinalizeSettingsCommand(ctx context.Context, sessionID, commandID string, finalize store.SettingsCommandFinalize) (store.SettingsCommand, error)

func (*Store) History added in v0.1.7 

func (s *Store) History(ctx context.Context, sessionID string, beforeSeq *int64, limit int) (store.HistoryPage, error)

func (*Store) InitializeAdapterConnection added in v0.1.7 

func (s *Store) InitializeAdapterConnection(ctx context.Context, request store.AdapterConnectionInitialize) (store.AdapterConnection, error)

func (*Store) IssueAdapterConnectionAuthorityReceipt added in v0.1.7 

func (s *Store) IssueAdapterConnectionAuthorityReceipt(ctx context.Context, sessionID string, admission store.AdapterConnectionAdmission, writer store.SettingsWriter) (store.ConnectionAuthorityReceipt, error)

func (*Store) LatestSeq 

func (s *Store) LatestSeq(ctx context.Context, sessionID string) (int64, error)

func (*Store) ListPendingCommands added in v0.1.7 

func (s *Store) ListPendingCommands(ctx context.Context, sessionID string, authority store.CommandAuthority) ([]store.PendingCommand, error)

func (*Store) PendingFileReferenceCommands added in v0.1.7 

func (s *Store) PendingFileReferenceCommands(ctx context.Context, sessionID string) ([]store.FileReferenceCommand, error)

func (*Store) PendingRunControls added in v0.1.7 

func (s *Store) PendingRunControls(ctx context.Context, sessionID string) ([]store.RunControlReservation, error)

func (*Store) PendingSettingsCommands added in v0.1.7 

func (s *Store) PendingSettingsCommands(ctx context.Context, sessionID string) ([]store.SettingsCommand, error)

func (*Store) PrepareAdapterCredentialRotation added in v0.1.7 

func (s *Store) PrepareAdapterCredentialRotation(ctx context.Context, sessionID string, rotation store.AdapterCredentialRotation) (store.AdapterConnection, error)

func (*Store) PublishFileReferenceCapability added in v0.1.7 

func (s *Store) PublishFileReferenceCapability(ctx context.Context, sessionID string, update store.FileReferenceCapabilityUpdate) (store.FileReferenceCapability, error)

func (*Store) PublishRunControlCapability added in v0.1.7 

func (s *Store) PublishRunControlCapability(ctx context.Context, sessionID string, update store.RunControlCapabilityUpdate) (store.RunControlCapability, error)

func (*Store) PublishSettingsCapability added in v0.1.7 

func (s *Store) PublishSettingsCapability(ctx context.Context, sessionID string, update store.SettingsCapabilityUpdate) (store.SettingsCapability, error)

func (*Store) QuarantineWorkspaceLease added in v0.1.7 

func (s *Store) QuarantineWorkspaceLease(ctx context.Context, key store.WorkspaceLeaseKey, expectedVersion int64) (store.WorkspaceLease, error)

func (*Store) RecordProviderStartAdmission added in v0.1.7 

func (s *Store) RecordProviderStartAdmission(ctx context.Context, request store.ProviderStartAdmission) (store.WorkspaceLease, error)

RecordProviderStartAdmission uses one PostgreSQL transaction to lock the exact Adapter connection and its sole reserved writer lease before recording start_received. The Hub derives the lease from durable tuple truth; the Adapter cannot nominate a workspace or bypass quarantine.

func (*Store) RecordWorkspaceStartReceived added in v0.1.7 

func (s *Store) RecordWorkspaceStartReceived(ctx context.Context, key store.WorkspaceLeaseKey, expectedVersion int64, owner store.WorkspaceLeaseOwner) (store.WorkspaceLease, error)

func (*Store) RecoverFileReferenceCommand added in v0.1.7 

func (s *Store) RecoverFileReferenceCommand(ctx context.Context, sessionID, commandID, reason string) (store.FileReferenceCommand, error)

func (*Store) RecoverRunControl added in v0.1.7 

func (s *Store) RecoverRunControl(ctx context.Context, sessionID, commandID, reason string) (store.RunControlReservation, error)

func (*Store) RecoverSettingsCommand added in v0.1.7 

func (s *Store) RecoverSettingsCommand(ctx context.Context, sessionID, commandID string, priorWriter store.SettingsWriter) (store.SettingsCommand, error)

func (*Store) RefreshAdapterCredentialBeforeHello added in v0.1.7 

func (s *Store) RefreshAdapterCredentialBeforeHello(ctx context.Context, sessionID string, refresh store.AdapterCredentialPreHelloRefresh) (store.AdapterConnection, error)

func (*Store) ReleaseWorkspaceLeaseAfterQuiescence added in v0.1.7 

func (s *Store) ReleaseWorkspaceLeaseAfterQuiescence(ctx context.Context, key store.WorkspaceLeaseKey, expectedVersion int64, owner store.WorkspaceLeaseOwner) (store.WorkspaceLease, error)

func (*Store) Replay 

func (s *Store) Replay(ctx context.Context, sessionID string, afterSeq int64, fn func(store.Event) error) (err error)

func (*Store) ReserveWorkspaceLease added in v0.1.7 

func (s *Store) ReserveWorkspaceLease(ctx context.Context, reserve store.WorkspaceLeaseReserve) (store.WorkspaceLease, error)

func (*Store) ResolvePendingCommand added in v0.1.7 

func (s *Store) ResolvePendingCommand(ctx context.Context, sessionID string, authority store.CommandAuthority, commandID string, status store.PendingCommandStatus) (store.PendingCommand, error)

func (*Store) ResolvePendingCommandUnknown added in v0.1.7 

func (s *Store) ResolvePendingCommandUnknown(ctx context.Context, sessionID string, commandID string) (store.PendingCommand, error)

func (*Store) RunAttentionBackfill added in v0.1.7 

func (s *Store) RunAttentionBackfill(ctx context.Context, checkpoints AttentionBackfillCheckpointStore, batchSize int) (AttentionBackfillResult, error)

func (*Store) RunControl added in v0.1.7 

func (s *Store) RunControl(ctx context.Context, sessionID, commandID string) (store.RunControlReservation, error)

func (*Store) RunControlFinalize added in v0.1.7 

func (s *Store) RunControlFinalize(ctx context.Context, sessionID, commandID string, finalize store.RunControlFinalize) (store.RunControlReservation, error)

func (*Store) RunControlReserve added in v0.1.7 

func (s *Store) RunControlReserve(ctx context.Context, sessionID string, request store.RunControlRequest) (store.RunControlReserve, error)

func (*Store) SessionAdmissionTruth added in v0.1.7 

func (s *Store) SessionAdmissionTruth(ctx context.Context, sessionID string) (store.SessionAdmissionTruth, error)

func (*Store) SettingsCommand added in v0.1.7 

func (s *Store) SettingsCommand(ctx context.Context, sessionID, commandID string) (store.SettingsCommand, error)

func (*Store) SettingsCommandReserve added in v0.1.7 

func (s *Store) SettingsCommandReserve(ctx context.Context, sessionID string, request store.SettingsCommandRequest) (store.SettingsCommandReserve, error)

func (*Store) TerminateAdapterConnectionBeforeHello added in v0.1.7 

func (s *Store) TerminateAdapterConnectionBeforeHello(ctx context.Context, sessionID string, termination store.AdapterConnectionPreHelloTermination) (store.AdapterConnection, error)

func (*Store) UpdateAttachment added in v0.1.7 

func (s *Store) UpdateAttachment(ctx context.Context, attachID string, expectedVersion int64, update store.AttachmentUpdate) (store.AttachmentMutation, error)

func (*Store) ValidateAdapterAdmission added in v0.1.7 

func (s *Store) ValidateAdapterAdmission(ctx context.Context, sessionID string, admission store.AdapterConnectionAdmission) (store.AdapterConnection, error)

func (*Store) ValidateAdapterEffectAdmission added in v0.1.7 

func (s *Store) ValidateAdapterEffectAdmission(ctx context.Context, sessionID string, admission store.AdapterConnectionAdmission) (store.AdapterConnection, error)

func (*Store) ValidateWarmAttachTargetActivation added in v0.1.7 

func (s *Store) ValidateWarmAttachTargetActivation(ctx context.Context, sessionID string, activation store.WarmAttachTargetActivation) error

func (*Store) WithAdapterConnectionTransaction added in v0.1.7 

func (s *Store) WithAdapterConnectionTransaction(ctx context.Context, fn func(store.AdapterConnectionStore) error) error

func (*Store) WithProviderStartAdmission added in v0.1.7 

func (s *Store) WithProviderStartAdmission(ctx context.Context, request store.ProviderStartAdmission, callback func(context.Context) error) (store.WorkspaceLease, error)

WithProviderStartAdmission keeps the Store's connection and writer locks through the physical start proof. Attached targets additionally hold their reserved WorkspaceLease; standalone bootstrap Sessions have no workspace state to transition.

func (*Store) WorkspaceLease added in v0.1.7 

func (s *Store) WorkspaceLease(ctx context.Context, key store.WorkspaceLeaseKey) (store.WorkspaceLease, error)

Source Files

View all Source files

Directories

Path Synopsis
internal
db

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.