From a8f1fdae71613ac03b535260f94afac006d9691a Mon Sep 17 00:00:00 2001 From: Xinhao Xu <84456268+xxhZs@users.noreply.github.com> Date: Fri, 31 Jul 2026 13:14:36 +0800 Subject: [PATCH] feat(storage): enforce session commit protocol --- README.md | 11 +- docs/session-commit-protocol.md | 85 ++ internal/resource/session.go | 138 +++ internal/store/postgres/migrate.go | 197 +++- .../000002_session_commit_protocol.down.sql | 5 + .../000002_session_commit_protocol.up.sql | 130 +++ internal/store/postgres/session.go | 1019 +++++++++++++++++ .../postgres/session_integration_test.go | 451 ++++++++ internal/store/postgres/session_test.go | 122 ++ internal/store/postgres/store.go | 16 +- internal/store/postgres/store_test.go | 27 +- internal/store/session.go | 72 ++ 12 files changed, 2244 insertions(+), 29 deletions(-) create mode 100644 docs/session-commit-protocol.md create mode 100644 internal/resource/session.go create mode 100644 internal/store/postgres/migrations/000002_session_commit_protocol.down.sql create mode 100644 internal/store/postgres/migrations/000002_session_commit_protocol.up.sql create mode 100644 internal/store/postgres/session.go create mode 100644 internal/store/postgres/session_integration_test.go create mode 100644 internal/store/postgres/session_test.go create mode 100644 internal/store/session.go diff --git a/README.md b/README.md index b654fea..280cf72 100644 --- a/README.md +++ b/README.md @@ -11,9 +11,12 @@ one Activation at a time. ## Status The repository contains the PostgreSQL-authoritative SandboxAttempt reconcile -slice: atomic resource/change/outbox persistence, at-least-once Hatchet -dispatch, tenant-scoped workers, and OpenSandbox adoption after ambiguous -provider outcomes. The broader Cloud Agent API remains under development. +slice and the CloudSession commit consistency boundary: atomic +resource/change/outbox persistence, at-least-once Hatchet dispatch, +tenant-scoped workers, OpenSandbox adoption after ambiguous provider outcomes, +Activation admission, Session lease fencing, immutable BundleRevision +metadata, and atomic commit/publication visibility. The broader Cloud Agent API +and execution orchestration remain under development. ## Layout @@ -40,6 +43,8 @@ The first Maka Graph implementation plan is in The PostgreSQL/Hatchet reconciliation boundary and downstream package ownership are specified in [`docs/reconcile-workflow-contract.md`](docs/reconcile-workflow-contract.md). +The implemented Session consistency boundary is described in +[`docs/session-commit-protocol.md`](docs/session-commit-protocol.md). ## Run diff --git a/docs/session-commit-protocol.md b/docs/session-commit-protocol.md new file mode 100644 index 0000000..0729752 --- /dev/null +++ b/docs/session-commit-protocol.md @@ -0,0 +1,85 @@ +# Session commit protocol + +This slice implements the PostgreSQL consistency boundary for one Cloud +Session turn. It does not run Maka or move Session Bundle bytes. + +## Resource relationship + +```text +CloudSession + +-- Activation (one admitted stimulus) + | +-- SandboxAttempt (one replaceable execution attempt) + +-- SessionLease (one current execution fence) + +-- BundleRevision (immutable committed checkpoint metadata) +``` + +An Agent is referenced configuration, not a process. Different Sessions that +reference the same Agent may execute concurrently. Activations within one +Session are serialized. + +## Server-owned admission state + +Callers provide Session identity, Agent identity, and desired lifecycle, but +cannot choose initial metadata or status. The repository assigns generation 1, +resource version 1, and either `Ready` or `Suspended`. + +Activation admission likewise rejects caller-owned status, resource version, +generation, finalizers, and Agent generation. It snapshots the Session's Agent +generation and creates a `Pending` Activation. The admission key is +`(tenant, cloud_session_id, activation_id)`. + +Idempotency hashes canonicalize JSON object ordering with `json.Number`, so +integers beyond IEEE-754 precision are not collapsed. An exact admission replay +returns the existing Activation; reusing the key with different input is a +non-retryable duplicate. + +## Claiming and fencing + +The oldest pending Activation is the only one allowed to claim an idle +Session. `AcquireSessionLease` records the holder, Activation, monotonically +increasing epoch, and expiry. It also snapshots the Session head into the +Activation and marks both resources active in the same transaction. + +The database clock is authoritative for expiry. A live lease cannot be stolen. +After expiry, the same claimed Activation may be recovered by another worker +with a higher epoch. Lease renewal and commit validate the holder, Activation, +epoch, resource version, and expiry. + +## Commit transaction + +The candidate Bundle bytes must already exist in object storage. A successful +`CommitActivation` performs one metadata transaction: + +1. lock the CloudSession, Activation, and SessionLease rows; +2. verify the expected Session head and lease epoch; +3. insert immutable BundleRevision metadata for `head + 1`; +4. advance the Session head and clear its active Activation; +5. record the committed Activation outcome; +6. insert the reply into `publication_outbox`; +7. expire the lease; +8. append ordered resource changes for the Session, Activation, and revision. + +No reply becomes visible without its committed Session head. An exact commit +replay returns the existing revision and outbox identity. A replay that changes +the expected parent, bundle, outcome, publication, or lease epoch is rejected. + +Resource-change allocation takes a tenant-scoped transaction advisory lock. +This prevents a watcher from advancing past a lower change ID that commits +later. + +## Migrations + +Migrations are embedded, ordered by numeric version, and recorded in +`maka_cloud_schema_migrations` under one transaction advisory lock. The runner +recognizes a complete database created by the original version-1 runner, +records that baseline, and applies version 2. A partial legacy schema is +rejected instead of being guessed into a valid state. + +## Deliberate boundaries + +- Bundle bytes remain an object-storage concern. +- Provider, object-storage, Maka, and messaging calls never run inside the + metadata transaction. +- The repository does not choose queue or workflow technology. +- Publication delivery, Activation state-machine orchestration, cancellation, + and garbage collection remain follow-up slices. diff --git a/internal/resource/session.go b/internal/resource/session.go new file mode 100644 index 0000000..6bea292 --- /dev/null +++ b/internal/resource/session.go @@ -0,0 +1,138 @@ +package resource + +import ( + "encoding/json" + "time" +) + +type CloudSessionLifecycle string + +const ( + CloudSessionLifecycleActive CloudSessionLifecycle = "Active" + CloudSessionLifecycleSuspended CloudSessionLifecycle = "Suspended" +) + +type CloudSessionPhase string + +const ( + CloudSessionReady CloudSessionPhase = "Ready" + CloudSessionBusy CloudSessionPhase = "Busy" + CloudSessionSuspended CloudSessionPhase = "Suspended" + CloudSessionTerminating CloudSessionPhase = "Terminating" +) + +type CloudSessionSpec struct { + AgentRef string + AgentGeneration int64 + Lifecycle CloudSessionLifecycle +} + +type CloudSessionStatus struct { + ObservedGeneration int64 + HeadRevision int64 + LastCommittedActivationID string + ActiveActivationID string + Phase CloudSessionPhase +} + +type CloudSession struct { + Metadata Metadata + Spec CloudSessionSpec + Status CloudSessionStatus +} + +type ActivationStimulusType string + +const ( + ActivationStimulusMessage ActivationStimulusType = "message" + ActivationStimulusSchedule ActivationStimulusType = "schedule" + ActivationStimulusSystem ActivationStimulusType = "system" +) + +type ActivationStimulus struct { + Type ActivationStimulusType + Payload json.RawMessage +} + +type ActivationSpec struct { + CloudSessionID string + ActivationID string + AgentGeneration int64 + Stimulus ActivationStimulus + Deadline *time.Time +} + +type ActivationPhase string + +const ( + ActivationPending ActivationPhase = "Pending" + ActivationClaimed ActivationPhase = "Claimed" + ActivationScheduled ActivationPhase = "Scheduled" + ActivationHydrating ActivationPhase = "Hydrating" + ActivationRunning ActivationPhase = "Running" + ActivationPacking ActivationPhase = "Packing" + ActivationCommitting ActivationPhase = "Committing" + ActivationPublishing ActivationPhase = "Publishing" + ActivationSucceeded ActivationPhase = "Succeeded" + ActivationRetrying ActivationPhase = "RetryPending" + ActivationBlocked ActivationPhase = "Blocked" + ActivationFailed ActivationPhase = "Failed" + ActivationCancelled ActivationPhase = "Cancelled" +) + +type ActivationOutcomeStatus string + +const ( + ActivationOutcomeCompleted ActivationOutcomeStatus = "completed" + ActivationOutcomeBlocked ActivationOutcomeStatus = "blocked" + ActivationOutcomeRetryableFailure ActivationOutcomeStatus = "retryable_failure" + ActivationOutcomeFatalFailure ActivationOutcomeStatus = "fatal_failure" +) + +type ActivationOutcome struct { + Status ActivationOutcomeStatus + Response json.RawMessage + Reason string +} + +type ActivationStatus struct { + ObservedGeneration int64 + Phase ActivationPhase + Attempt int + BaseRevision *int64 + SandboxAttemptID string + LeaseEpoch *int64 + CommittedRevision *int64 + Outcome *ActivationOutcome +} + +type Activation struct { + Metadata Metadata + Spec ActivationSpec + Status ActivationStatus +} + +type SessionLease struct { + TenantID string + CloudSessionID string + HolderID string + ActivationID string + Epoch int64 + ResourceVersion string + AcquiredAt time.Time + RenewedAt time.Time + ExpiresAt time.Time +} + +type BundleRevision struct { + TenantID string + CloudSessionID string + Revision int64 + ParentRevision int64 + BundleRef string + TransportDigest string + PayloadDigest string + SizeBytes int64 + CreatedByActivationID string + CreatedAt time.Time +} diff --git a/internal/store/postgres/migrate.go b/internal/store/postgres/migrate.go index 69ecec6..ace941e 100644 --- a/internal/store/postgres/migrate.go +++ b/internal/store/postgres/migrate.go @@ -4,6 +4,10 @@ import ( "context" "embed" "fmt" + "io/fs" + "sort" + "strconv" + "strings" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -20,32 +24,36 @@ const ( MigrationDown MigrationDirection = "down" ) -// MigrationSQL returns the embedded migration for explicit tooling and tests. +const migrationLockID int64 = 0x4d414b41434c4f55 + +type migration struct { + version int64 + name string + sql string +} + +// MigrationSQL returns all embedded migrations in execution order. ApplyMigration +// should be used for real databases because it tracks individual versions. func MigrationSQL(direction MigrationDirection) (string, error) { - var name string - switch direction { - case MigrationUp: - name = "migrations/000001_sandbox_reconcile.up.sql" - case MigrationDown: - name = "migrations/000001_sandbox_reconcile.down.sql" - default: - return "", fmt.Errorf("unknown migration direction %q", direction) - } - b, err := migrations.ReadFile(name) + list, err := migrationList(direction) if err != nil { - return "", fmt.Errorf("read embedded migration: %w", err) + return "", err } - return string(b), nil + var out strings.Builder + for _, item := range list { + fmt.Fprintf(&out, "-- %s\n%s\n", item.name, item.sql) + } + return out.String(), nil } -// ApplyMigration applies the single embedded migration in one transaction. -// Applications do not call this during startup; it is intended for explicit -// deployment tooling. +// ApplyMigration applies pending embedded versions under one transaction-level +// advisory lock. It recognizes databases created by the original single-file +// runner and records version 1 before applying later migrations. func ApplyMigration(ctx context.Context, pool *pgxpool.Pool, direction MigrationDirection) error { if pool == nil { return &store.Error{Kind: store.ErrorInvalid, Operation: "migrate", Err: fmt.Errorf("nil PostgreSQL pool")} } - sql, err := MigrationSQL(direction) + list, err := migrationList(direction) if err != nil { return &store.Error{Kind: store.ErrorInvalid, Operation: "migrate", Err: err} } @@ -54,11 +62,162 @@ func ApplyMigration(ctx context.Context, pool *pgxpool.Pool, direction Migration return mapError("begin migration", err) } defer tx.Rollback(context.Background()) //nolint:errcheck - if _, err := tx.Exec(ctx, sql, pgx.QueryExecModeSimpleProtocol); err != nil { - return mapError("apply migration", err) + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, migrationLockID); err != nil { + return mapError("lock migrations", err) + } + if _, err := tx.Exec(ctx, ` + CREATE TABLE IF NOT EXISTS maka_cloud_schema_migrations ( + version bigint PRIMARY KEY CHECK (version > 0), + name text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT clock_timestamp() + )`); err != nil { + return mapError("create migration ledger", err) + } + if err := bootstrapLegacyBaseline(ctx, tx); err != nil { + return err + } + if err := validateMigrationLedger(ctx, tx, list); err != nil { + return err + } + for _, item := range list { + var applied bool + if err := tx.QueryRow(ctx, `SELECT EXISTS ( + SELECT 1 FROM maka_cloud_schema_migrations WHERE version=$1 + )`, item.version).Scan(&applied); err != nil { + return mapError("read migration ledger", err) + } + if direction == MigrationUp && applied { + continue + } + if direction == MigrationDown && !applied { + continue + } + if _, err := tx.Exec(ctx, item.sql, pgx.QueryExecModeSimpleProtocol); err != nil { + return mapError("apply migration "+item.name, err) + } + if direction == MigrationUp { + if _, err := tx.Exec(ctx, ` + INSERT INTO maka_cloud_schema_migrations(version,name) + VALUES($1,$2)`, item.version, item.name); err != nil { + return mapError("record migration "+item.name, err) + } + } else if _, err := tx.Exec(ctx, ` + DELETE FROM maka_cloud_schema_migrations WHERE version=$1`, item.version); err != nil { + return mapError("remove migration "+item.name, err) + } + } + if direction == MigrationDown { + if _, err := tx.Exec(ctx, `DROP TABLE maka_cloud_schema_migrations`); err != nil { + return mapError("drop migration ledger", err) + } } if err := tx.Commit(ctx); err != nil { return mapError("commit migration", err) } return nil } + +func migrationList(direction MigrationDirection) ([]migration, error) { + if direction != MigrationUp && direction != MigrationDown { + return nil, fmt.Errorf("unknown migration direction %q", direction) + } + suffix := "." + string(direction) + ".sql" + names, err := fs.Glob(migrations, "migrations/*"+suffix) + if err != nil { + return nil, fmt.Errorf("list embedded migrations: %w", err) + } + sort.Strings(names) + if direction == MigrationDown { + for left, right := 0, len(names)-1; left < right; left, right = left+1, right-1 { + names[left], names[right] = names[right], names[left] + } + } + out := make([]migration, 0, len(names)) + versions := make(map[int64]string, len(names)) + for _, name := range names { + base := strings.TrimPrefix(name, "migrations/") + versionText, _, ok := strings.Cut(base, "_") + if !ok { + return nil, fmt.Errorf("invalid migration filename %q", name) + } + version, err := strconv.ParseInt(versionText, 10, 64) + if err != nil || version <= 0 { + return nil, fmt.Errorf("invalid migration version in %q", name) + } + if previous, exists := versions[version]; exists { + return nil, fmt.Errorf("duplicate migration version %d in %q and %q", version, previous, name) + } + versions[version] = name + body, err := migrations.ReadFile(name) + if err != nil { + return nil, fmt.Errorf("read embedded migration %q: %w", name, err) + } + out = append(out, migration{version: version, name: base, sql: string(body)}) + } + return out, nil +} + +func bootstrapLegacyBaseline(ctx context.Context, tx pgx.Tx) error { + var count int + if err := tx.QueryRow(ctx, `SELECT count(*) FROM maka_cloud_schema_migrations`).Scan(&count); err != nil { + return mapError("inspect migration ledger", err) + } + if count != 0 { + return nil + } + var attemptsExist, changesExist, outboxExists bool + if err := tx.QueryRow(ctx, ` + SELECT to_regclass('sandbox_attempts') IS NOT NULL, + to_regclass('resource_changes') IS NOT NULL, + to_regclass('reconcile_outbox') IS NOT NULL + `).Scan(&attemptsExist, &changesExist, &outboxExists); err != nil { + return mapError("inspect legacy migration", err) + } + if !attemptsExist && !changesExist && !outboxExists { + return nil + } + if !attemptsExist || !changesExist || !outboxExists { + return invalid("bootstrap legacy migration", "partial version 1 schema") + } + if _, err := tx.Exec(ctx, ` + INSERT INTO maka_cloud_schema_migrations(version,name) + VALUES(1,'000001_sandbox_reconcile.up.sql')`); err != nil { + return mapError("record legacy baseline", err) + } + return nil +} + +func validateMigrationLedger(ctx context.Context, tx pgx.Tx, available []migration) error { + known := make(map[int64]bool, len(available)) + for _, item := range available { + known[item.version] = true + } + rows, err := tx.Query(ctx, `SELECT version FROM maka_cloud_schema_migrations ORDER BY version`) + if err != nil { + return mapError("read migration ledger", err) + } + defer rows.Close() + for rows.Next() { + var version int64 + if err := rows.Scan(&version); err != nil { + return mapError("scan migration ledger", err) + } + if !known[version] { + return invalid("validate migration ledger", fmt.Sprintf("applied migration %d is unavailable for %s", version, availableDirection(available))) + } + } + if err := rows.Err(); err != nil { + return mapError("read migration ledger", err) + } + return nil +} + +func availableDirection(list []migration) string { + if len(list) == 0 { + return "requested direction" + } + if strings.Contains(list[0].name, ".down.sql") { + return "down migration" + } + return "up migration" +} diff --git a/internal/store/postgres/migrations/000002_session_commit_protocol.down.sql b/internal/store/postgres/migrations/000002_session_commit_protocol.down.sql new file mode 100644 index 0000000..683b1a4 --- /dev/null +++ b/internal/store/postgres/migrations/000002_session_commit_protocol.down.sql @@ -0,0 +1,5 @@ +DROP TABLE IF EXISTS publication_outbox; +DROP TABLE IF EXISTS bundle_revisions; +DROP TABLE IF EXISTS session_leases; +DROP TABLE IF EXISTS activations; +DROP TABLE IF EXISTS cloud_sessions; diff --git a/internal/store/postgres/migrations/000002_session_commit_protocol.up.sql b/internal/store/postgres/migrations/000002_session_commit_protocol.up.sql new file mode 100644 index 0000000..87404a8 --- /dev/null +++ b/internal/store/postgres/migrations/000002_session_commit_protocol.up.sql @@ -0,0 +1,130 @@ +CREATE TABLE cloud_sessions ( + tenant_id text NOT NULL CHECK (tenant_id <> ''), + cloud_session_id text NOT NULL CHECK (cloud_session_id <> ''), + resource_version bigint NOT NULL CHECK (resource_version > 0), + generation bigint NOT NULL CHECK (generation > 0), + metadata jsonb NOT NULL, + spec jsonb NOT NULL, + status jsonb NOT NULL, + lifecycle text NOT NULL CHECK (lifecycle IN ('Active', 'Suspended')), + phase text NOT NULL CHECK (phase IN ('Ready', 'Busy', 'Suspended', 'Terminating')), + head_revision bigint NOT NULL CHECK (head_revision >= 0), + active_activation_id text, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + updated_at timestamptz NOT NULL DEFAULT clock_timestamp(), + CHECK (metadata->>'TenantID' = tenant_id), + CHECK (metadata->>'ID' = cloud_session_id), + CHECK (metadata->>'ResourceVersion' = resource_version::text), + CHECK ((metadata->>'Generation')::bigint = generation), + CHECK (spec->>'Lifecycle' = lifecycle), + CHECK (status->>'Phase' = phase), + CHECK ((status->>'HeadRevision')::bigint = head_revision), + CHECK (COALESCE(NULLIF(status->>'ActiveActivationID',''),'') = + COALESCE(active_activation_id,'')), + CHECK (phase <> 'Busy' OR active_activation_id IS NOT NULL), + CHECK (active_activation_id IS NULL OR phase IN ('Busy', 'Terminating')), + PRIMARY KEY (tenant_id, cloud_session_id) +); + +CREATE TABLE activations ( + tenant_id text NOT NULL CHECK (tenant_id <> ''), + activation_resource_id text NOT NULL CHECK (activation_resource_id <> ''), + cloud_session_id text NOT NULL CHECK (cloud_session_id <> ''), + activation_id text NOT NULL CHECK (activation_id <> ''), + request_hash bytea NOT NULL, + resource_version bigint NOT NULL CHECK (resource_version > 0), + generation bigint NOT NULL CHECK (generation > 0), + metadata jsonb NOT NULL, + spec jsonb NOT NULL, + status jsonb NOT NULL, + phase text NOT NULL CHECK ( + phase IN ( + 'Pending', 'Claimed', 'Scheduled', 'Hydrating', 'Running', + 'Packing', 'Committing', 'Publishing', 'Succeeded', + 'RetryPending', 'Blocked', 'Failed', 'Cancelled' + ) + ), + base_revision bigint, + lease_epoch bigint, + committed_revision bigint, + outcome jsonb, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + updated_at timestamptz NOT NULL DEFAULT clock_timestamp(), + CHECK (metadata->>'TenantID' = tenant_id), + CHECK (metadata->>'ID' = activation_resource_id), + CHECK (metadata->>'ResourceVersion' = resource_version::text), + CHECK ((metadata->>'Generation')::bigint = generation), + CHECK (spec->>'CloudSessionID' = cloud_session_id), + CHECK (spec->>'ActivationID' = activation_id), + CHECK (status->>'Phase' = phase), + PRIMARY KEY (tenant_id, activation_resource_id), + UNIQUE (tenant_id, cloud_session_id, activation_id), + FOREIGN KEY (tenant_id, cloud_session_id) + REFERENCES cloud_sessions (tenant_id, cloud_session_id) +); + +CREATE INDEX activations_work_idx + ON activations (tenant_id, cloud_session_id, created_at, activation_resource_id) + WHERE phase IN ('Pending', 'RetryPending'); + +CREATE TABLE session_leases ( + tenant_id text NOT NULL CHECK (tenant_id <> ''), + cloud_session_id text NOT NULL CHECK (cloud_session_id <> ''), + holder_id text NOT NULL CHECK (holder_id <> ''), + activation_id text NOT NULL CHECK (activation_id <> ''), + epoch bigint NOT NULL CHECK (epoch > 0), + resource_version bigint NOT NULL CHECK (resource_version > 0), + acquired_at timestamptz NOT NULL, + renewed_at timestamptz NOT NULL, + expires_at timestamptz NOT NULL, + PRIMARY KEY (tenant_id, cloud_session_id), + FOREIGN KEY (tenant_id, cloud_session_id) + REFERENCES cloud_sessions (tenant_id, cloud_session_id), + FOREIGN KEY (tenant_id, cloud_session_id, activation_id) + REFERENCES activations (tenant_id, cloud_session_id, activation_id) +); + +CREATE INDEX session_leases_expiry_idx ON session_leases (expires_at); + +CREATE TABLE bundle_revisions ( + tenant_id text NOT NULL CHECK (tenant_id <> ''), + cloud_session_id text NOT NULL CHECK (cloud_session_id <> ''), + revision bigint NOT NULL CHECK (revision > 0), + parent_revision bigint NOT NULL CHECK (parent_revision >= 0), + bundle_ref text NOT NULL CHECK (bundle_ref <> ''), + transport_digest text NOT NULL CHECK (transport_digest <> ''), + payload_digest text NOT NULL CHECK (payload_digest <> ''), + size_bytes bigint NOT NULL CHECK (size_bytes >= 0), + created_by_activation_id text NOT NULL CHECK (created_by_activation_id <> ''), + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + PRIMARY KEY (tenant_id, cloud_session_id, revision), + UNIQUE (tenant_id, cloud_session_id, created_by_activation_id), + FOREIGN KEY (tenant_id, cloud_session_id) + REFERENCES cloud_sessions (tenant_id, cloud_session_id), + FOREIGN KEY (tenant_id, cloud_session_id, created_by_activation_id) + REFERENCES activations (tenant_id, cloud_session_id, activation_id) +); + +CREATE TABLE publication_outbox ( + outbox_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + tenant_id text NOT NULL CHECK (tenant_id <> ''), + cloud_session_id text NOT NULL CHECK (cloud_session_id <> ''), + activation_id text NOT NULL CHECK (activation_id <> ''), + destination text NOT NULL CHECK (destination <> ''), + payload jsonb NOT NULL, + attempts integer NOT NULL DEFAULT 0 CHECK (attempts >= 0), + available_at timestamptz NOT NULL DEFAULT clock_timestamp(), + claimed_by text, + claimed_at timestamptz, + delivered_at timestamptz, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + UNIQUE (tenant_id, cloud_session_id, activation_id), + FOREIGN KEY (tenant_id, cloud_session_id) + REFERENCES cloud_sessions (tenant_id, cloud_session_id), + FOREIGN KEY (tenant_id, cloud_session_id, activation_id) + REFERENCES activations (tenant_id, cloud_session_id, activation_id) +); + +CREATE INDEX publication_outbox_pending_idx + ON publication_outbox (available_at, outbox_id) + WHERE delivered_at IS NULL; diff --git a/internal/store/postgres/session.go b/internal/store/postgres/session.go new file mode 100644 index 0000000..b6b6919 --- /dev/null +++ b/internal/store/postgres/session.go @@ -0,0 +1,1019 @@ +package postgres + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "strconv" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/maka-agent/maka-agent-cloud/internal/resource" + storeport "github.com/maka-agent/maka-agent-cloud/internal/store" +) + +type sessionStore struct{ authority *Store } + +var _ storeport.SessionRepository = (*sessionStore)(nil) + +func (s *sessionStore) CreateCloudSession(ctx context.Context, input resource.CloudSession) (resource.CloudSession, error) { + normalized, err := normalizeNewCloudSession(input) + if err != nil { + return resource.CloudSession{}, err + } + var result resource.CloudSession + err = s.authority.transact(ctx, func(tx *transaction) error { + repo := newSessionTx(tx) + result, err = repo.createCloudSession(ctx, normalized) + return err + }) + return result, err +} + +func (s *sessionStore) GetCloudSession(ctx context.Context, tenantID, sessionID string) (resource.CloudSession, error) { + return getCloudSession(ctx, s.authority.pool, tenantID, sessionID, false) +} + +func (s *sessionStore) AdmitActivation(ctx context.Context, input resource.Activation) (resource.Activation, bool, error) { + normalized, err := normalizeNewActivation(input) + if err != nil { + return resource.Activation{}, false, err + } + var result resource.Activation + var created bool + err = s.authority.transact(ctx, func(tx *transaction) error { + repo := newSessionTx(tx) + result, created, err = repo.admitActivation(ctx, normalized) + return err + }) + return result, created, err +} + +func (s *sessionStore) GetActivation(ctx context.Context, tenantID, sessionID, activationID string) (resource.Activation, error) { + result, _, err := getActivation(ctx, s.authority.pool, tenantID, sessionID, activationID, false) + return result, err +} + +func (s *sessionStore) AcquireSessionLease(ctx context.Context, tenantID, sessionID, activationID, holderID string, ttl time.Duration) (resource.SessionLease, bool, error) { + if !validIDs(tenantID, sessionID, activationID, holderID) || ttl <= 0 { + return resource.SessionLease{}, false, invalid("acquire session lease", "invalid identity or TTL") + } + var result resource.SessionLease + var acquired bool + var err error + err = s.authority.transact(ctx, func(tx *transaction) error { + result, acquired, err = newSessionTx(tx).acquireSessionLease(ctx, tenantID, sessionID, activationID, holderID, ttl) + return err + }) + return result, acquired, err +} + +func (s *sessionStore) RenewSessionLease(ctx context.Context, lease resource.SessionLease, ttl time.Duration) (resource.SessionLease, error) { + if !validIDs(lease.TenantID, lease.CloudSessionID, lease.ActivationID, lease.HolderID, lease.ResourceVersion) || lease.Epoch <= 0 || ttl <= 0 { + return resource.SessionLease{}, invalid("renew session lease", "invalid lease identity, epoch, version, or TTL") + } + var result resource.SessionLease + var err error + err = s.authority.transact(ctx, func(tx *transaction) error { + result, err = newSessionTx(tx).renewSessionLease(ctx, lease, ttl) + return err + }) + return result, err +} + +func (s *sessionStore) CommitActivation(ctx context.Context, input storeport.CommitActivationRequest) (storeport.CommitActivationResult, error) { + normalized, err := normalizeCommitRequest(input) + if err != nil { + return storeport.CommitActivationResult{}, err + } + var result storeport.CommitActivationResult + err = s.authority.transact(ctx, func(tx *transaction) error { + result, err = newSessionTx(tx).commitActivation(ctx, normalized) + return err + }) + return result, err +} + +func (s *sessionStore) ListBundleRevisions(ctx context.Context, tenantID, sessionID string) ([]resource.BundleRevision, error) { + if !validIDs(tenantID, sessionID) { + return nil, invalid("list bundle revisions", "empty identity") + } + rows, err := s.authority.pool.Query(ctx, ` + SELECT tenant_id,cloud_session_id,revision,parent_revision,bundle_ref, + transport_digest,payload_digest,size_bytes,created_by_activation_id,created_at + FROM bundle_revisions + WHERE tenant_id=$1 AND cloud_session_id=$2 + ORDER BY revision`, tenantID, sessionID) + if err != nil { + return nil, mapError("list bundle revisions", err) + } + defer rows.Close() + var result []resource.BundleRevision + for rows.Next() { + var item resource.BundleRevision + if err := rows.Scan( + &item.TenantID, &item.CloudSessionID, &item.Revision, &item.ParentRevision, + &item.BundleRef, &item.TransportDigest, &item.PayloadDigest, &item.SizeBytes, + &item.CreatedByActivationID, &item.CreatedAt, + ); err != nil { + return nil, mapError("scan bundle revision", err) + } + item.CreatedAt = item.CreatedAt.UTC() + result = append(result, item) + } + if err := rows.Err(); err != nil { + return nil, mapError("list bundle revisions", err) + } + return result, nil +} + +func (s *sessionStore) ListPublicationOutbox(ctx context.Context, tenantID string, limit int) ([]storeport.PublicationOutboxEntry, error) { + if strings.TrimSpace(tenantID) == "" || limit <= 0 { + return nil, invalid("list publication outbox", "invalid tenant or limit") + } + rows, err := s.authority.pool.Query(ctx, ` + SELECT outbox_id,tenant_id,cloud_session_id,activation_id,destination,payload, + attempts,available_at,created_at,delivered_at + FROM publication_outbox + WHERE tenant_id=$1 + ORDER BY outbox_id + LIMIT $2`, tenantID, limit) + if err != nil { + return nil, mapError("list publication outbox", err) + } + defer rows.Close() + var result []storeport.PublicationOutboxEntry + for rows.Next() { + var item storeport.PublicationOutboxEntry + if err := rows.Scan( + &item.ID, &item.TenantID, &item.CloudSessionID, &item.ActivationID, + &item.Destination, &item.Payload, &item.Attempts, &item.AvailableAt, + &item.CreatedAt, &item.DeliveredAt, + ); err != nil { + return nil, mapError("scan publication outbox", err) + } + item.AvailableAt = item.AvailableAt.UTC() + item.CreatedAt = item.CreatedAt.UTC() + if item.DeliveredAt != nil { + value := item.DeliveredAt.UTC() + item.DeliveredAt = &value + } + result = append(result, item) + } + if err := rows.Err(); err != nil { + return nil, mapError("list publication outbox", err) + } + return result, nil +} + +type sessionTx struct { + q querier + state *txState +} + +func newSessionTx(tx *transaction) *sessionTx { + return &sessionTx{q: tx.attemptRepo.q, state: tx.attemptRepo.state} +} + +func (r *sessionTx) createCloudSession(ctx context.Context, input resource.CloudSession) (resource.CloudSession, error) { + metadata, spec, status, err := marshalCloudSession(input) + if err != nil { + return resource.CloudSession{}, err + } + result, err := scanCloudSession(r.q.QueryRow(ctx, ` + INSERT INTO cloud_sessions( + tenant_id,cloud_session_id,resource_version,generation,metadata,spec,status, + lifecycle,phase,head_revision,active_activation_id + ) VALUES($1,$2,1,1,$3,$4,$5,$6,$7,0,NULL) + RETURNING metadata,spec,status`, input.Metadata.TenantID, input.Metadata.ID, + metadata, spec, status, input.Spec.Lifecycle, input.Status.Phase), "create cloud session") + if err != nil { + return resource.CloudSession{}, err + } + if err := r.recordChange(ctx, result.Metadata.TenantID, storeport.ResourceKindCloudSession, result.Metadata.ID, result.Metadata.ResourceVersion); err != nil { + return resource.CloudSession{}, err + } + return result, nil +} + +func (r *sessionTx) admitActivation(ctx context.Context, input resource.Activation) (resource.Activation, bool, error) { + session, err := getCloudSession(ctx, r.q, input.Metadata.TenantID, input.Spec.CloudSessionID, true) + if err != nil { + return resource.Activation{}, false, err + } + if session.Metadata.DeletionTimestamp != nil || session.Spec.Lifecycle != resource.CloudSessionLifecycleActive { + return resource.Activation{}, false, invalid("admit activation", "cloud session is not active") + } + input.Spec.AgentGeneration = session.Spec.AgentGeneration + hash, err := activationRequestHash(input.Spec) + if err != nil { + return resource.Activation{}, false, err + } + metadata, spec, status, err := marshalActivation(input) + if err != nil { + return resource.Activation{}, false, err + } + result, _, err := scanActivation(r.q.QueryRow(ctx, ` + INSERT INTO activations( + tenant_id,activation_resource_id,cloud_session_id,activation_id,request_hash, + resource_version,generation,metadata,spec,status,phase + ) VALUES($1,$2,$3,$4,$5,1,1,$6,$7,$8,$9) + ON CONFLICT(tenant_id,cloud_session_id,activation_id) DO NOTHING + RETURNING metadata,spec,status,request_hash`, + input.Metadata.TenantID, input.Metadata.ID, input.Spec.CloudSessionID, + input.Spec.ActivationID, hash, metadata, spec, status, input.Status.Phase, + ), "admit activation") + if err == nil { + if err := r.recordChange(ctx, result.Metadata.TenantID, storeport.ResourceKindActivation, result.Metadata.ID, result.Metadata.ResourceVersion); err != nil { + return resource.Activation{}, false, err + } + return result, true, nil + } + if !storeport.IsKind(err, storeport.ErrorNotFound) { + return resource.Activation{}, false, err + } + existing, existingHash, err := getActivation(ctx, r.q, input.Metadata.TenantID, input.Spec.CloudSessionID, input.Spec.ActivationID, false) + if err != nil { + return resource.Activation{}, false, err + } + if !bytes.Equal(existingHash, hash) { + return resource.Activation{}, false, duplicate("admit activation", "activation key was reused with different input") + } + return existing, false, nil +} + +func (r *sessionTx) acquireSessionLease(ctx context.Context, tenantID, sessionID, activationID, holderID string, ttl time.Duration) (resource.SessionLease, bool, error) { + session, err := getCloudSession(ctx, r.q, tenantID, sessionID, true) + if err != nil { + return resource.SessionLease{}, false, err + } + if session.Metadata.DeletionTimestamp != nil || session.Spec.Lifecycle != resource.CloudSessionLifecycleActive { + return resource.SessionLease{}, false, invalid("acquire session lease", "cloud session is not active") + } + activation, _, err := getActivation(ctx, r.q, tenantID, sessionID, activationID, true) + if err != nil { + return resource.SessionLease{}, false, err + } + current, exists, err := getSessionLease(ctx, r.q, tenantID, sessionID, true) + if err != nil { + return resource.SessionLease{}, false, err + } + now, err := databaseNow(ctx, r.q) + if err != nil { + return resource.SessionLease{}, false, err + } + if exists && current.ExpiresAt.After(now) { + if current.ActivationID == activationID && current.HolderID == holderID && + session.Status.ActiveActivationID == activationID && activation.Status.Phase == resource.ActivationClaimed { + return current, false, nil + } + return resource.SessionLease{}, false, conflict("acquire session lease") + } + if session.Status.ActiveActivationID != "" && session.Status.ActiveActivationID != activationID { + return resource.SessionLease{}, false, conflict("acquire session lease") + } + if activation.Status.Phase != resource.ActivationPending && + activation.Status.Phase != resource.ActivationRetrying && + activation.Status.Phase != resource.ActivationClaimed { + return resource.SessionLease{}, false, invalid("acquire session lease", "activation is not claimable") + } + if activation.Status.Phase != resource.ActivationClaimed { + var oldest string + if err := r.q.QueryRow(ctx, ` + SELECT activation_id + FROM activations + WHERE tenant_id=$1 AND cloud_session_id=$2 AND phase IN ('Pending','RetryPending') + ORDER BY created_at,activation_resource_id + LIMIT 1 + FOR UPDATE`, tenantID, sessionID).Scan(&oldest); err != nil { + return resource.SessionLease{}, false, mapError("select oldest activation", err) + } + if oldest != activationID { + return resource.SessionLease{}, false, conflict("acquire session lease") + } + } + epoch, version := int64(1), int64(1) + if exists { + epoch = current.Epoch + 1 + parsed, err := strconv.ParseInt(current.ResourceVersion, 10, 64) + if err != nil { + return resource.SessionLease{}, false, invalidErr("acquire session lease", err) + } + version = parsed + 1 + } + expiresAt := now.Add(ttl) + var lease resource.SessionLease + if err := r.q.QueryRow(ctx, ` + INSERT INTO session_leases( + tenant_id,cloud_session_id,holder_id,activation_id,epoch,resource_version, + acquired_at,renewed_at,expires_at + ) VALUES($1,$2,$3,$4,$5,$6,$7,$7,$8) + ON CONFLICT(tenant_id,cloud_session_id) DO UPDATE SET + holder_id=EXCLUDED.holder_id, + activation_id=EXCLUDED.activation_id, + epoch=EXCLUDED.epoch, + resource_version=EXCLUDED.resource_version, + acquired_at=EXCLUDED.acquired_at, + renewed_at=EXCLUDED.renewed_at, + expires_at=EXCLUDED.expires_at + RETURNING tenant_id,cloud_session_id,holder_id,activation_id,epoch, + resource_version::text,acquired_at,renewed_at,expires_at`, + tenantID, sessionID, holderID, activationID, epoch, version, now, expiresAt, + ).Scan( + &lease.TenantID, &lease.CloudSessionID, &lease.HolderID, &lease.ActivationID, + &lease.Epoch, &lease.ResourceVersion, &lease.AcquiredAt, &lease.RenewedAt, + &lease.ExpiresAt, + ); err != nil { + return resource.SessionLease{}, false, mapError("persist session lease", err) + } + lease.AcquiredAt = lease.AcquiredAt.UTC() + lease.RenewedAt = lease.RenewedAt.UTC() + lease.ExpiresAt = lease.ExpiresAt.UTC() + + session.Status.ActiveActivationID = activationID + session.Status.Phase = resource.CloudSessionBusy + session.Status.ObservedGeneration = session.Metadata.Generation + session, err = r.updateCloudSession(ctx, session) + if err != nil { + return resource.SessionLease{}, false, err + } + baseRevision := session.Status.HeadRevision + activation.Status.Phase = resource.ActivationClaimed + activation.Status.ObservedGeneration = activation.Metadata.Generation + activation.Status.BaseRevision = &baseRevision + activation.Status.LeaseEpoch = &lease.Epoch + if _, err = r.updateActivation(ctx, activation); err != nil { + return resource.SessionLease{}, false, err + } + return lease, true, nil +} + +func (r *sessionTx) renewSessionLease(ctx context.Context, expected resource.SessionLease, ttl time.Duration) (resource.SessionLease, error) { + current, exists, err := getSessionLease(ctx, r.q, expected.TenantID, expected.CloudSessionID, true) + if err != nil { + return resource.SessionLease{}, err + } + if !exists { + return resource.SessionLease{}, &storeport.Error{Kind: storeport.ErrorNotFound, Operation: "renew session lease", Err: errors.New("lease not found")} + } + now, err := databaseNow(ctx, r.q) + if err != nil { + return resource.SessionLease{}, err + } + if current.HolderID != expected.HolderID || current.ActivationID != expected.ActivationID || + current.Epoch != expected.Epoch || current.ResourceVersion != expected.ResourceVersion || + !current.ExpiresAt.After(now) { + return resource.SessionLease{}, conflict("renew session lease") + } + version, err := strconv.ParseInt(current.ResourceVersion, 10, 64) + if err != nil { + return resource.SessionLease{}, invalidErr("renew session lease", err) + } + var renewed resource.SessionLease + err = r.q.QueryRow(ctx, ` + UPDATE session_leases + SET resource_version=$6,renewed_at=$7,expires_at=$8 + WHERE tenant_id=$1 AND cloud_session_id=$2 AND holder_id=$3 + AND activation_id=$4 AND epoch=$5 AND resource_version=$9 + AND expires_at>clock_timestamp() + RETURNING tenant_id,cloud_session_id,holder_id,activation_id,epoch, + resource_version::text,acquired_at,renewed_at,expires_at`, + expected.TenantID, expected.CloudSessionID, expected.HolderID, + expected.ActivationID, expected.Epoch, version+1, now, now.Add(ttl), version, + ).Scan( + &renewed.TenantID, &renewed.CloudSessionID, &renewed.HolderID, + &renewed.ActivationID, &renewed.Epoch, &renewed.ResourceVersion, + &renewed.AcquiredAt, &renewed.RenewedAt, &renewed.ExpiresAt, + ) + if err != nil { + mapped := mapError("renew session lease", err) + if storeport.IsKind(mapped, storeport.ErrorNotFound) { + return resource.SessionLease{}, conflict("renew session lease") + } + return resource.SessionLease{}, mapped + } + renewed.AcquiredAt = renewed.AcquiredAt.UTC() + renewed.RenewedAt = renewed.RenewedAt.UTC() + renewed.ExpiresAt = renewed.ExpiresAt.UTC() + return renewed, nil +} + +func (r *sessionTx) commitActivation(ctx context.Context, input storeport.CommitActivationRequest) (storeport.CommitActivationResult, error) { + session, err := getCloudSession(ctx, r.q, input.TenantID, input.CloudSessionID, true) + if err != nil { + return storeport.CommitActivationResult{}, err + } + activation, _, err := getActivation(ctx, r.q, input.TenantID, input.CloudSessionID, input.ActivationID, true) + if err != nil { + return storeport.CommitActivationResult{}, err + } + if activation.Status.CommittedRevision != nil { + return r.replayCommittedActivation(ctx, activation, input) + } + if session.Status.HeadRevision != input.ExpectedHeadRevision { + return storeport.CommitActivationResult{}, conflict("commit activation head") + } + if session.Status.ActiveActivationID != input.ActivationID || session.Status.Phase != resource.CloudSessionBusy { + return storeport.CommitActivationResult{}, invalid("commit activation", "cloud session is not owned by activation") + } + if activation.Status.Phase != resource.ActivationClaimed || + activation.Status.BaseRevision == nil || *activation.Status.BaseRevision != input.ExpectedHeadRevision || + activation.Status.LeaseEpoch == nil || *activation.Status.LeaseEpoch != input.LeaseEpoch { + return storeport.CommitActivationResult{}, invalid("commit activation", "activation claim does not match request") + } + lease, exists, err := getSessionLease(ctx, r.q, input.TenantID, input.CloudSessionID, true) + if err != nil { + return storeport.CommitActivationResult{}, err + } + now, err := databaseNow(ctx, r.q) + if err != nil { + return storeport.CommitActivationResult{}, err + } + if !exists || lease.HolderID != input.HolderID || lease.ActivationID != input.ActivationID || + lease.Epoch != input.LeaseEpoch || !lease.ExpiresAt.After(now) { + return storeport.CommitActivationResult{}, conflict("commit activation lease") + } + + nextRevision := input.ExpectedHeadRevision + 1 + if _, err := r.q.Exec(ctx, ` + INSERT INTO bundle_revisions( + tenant_id,cloud_session_id,revision,parent_revision,bundle_ref, + transport_digest,payload_digest,size_bytes,created_by_activation_id + ) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)`, + input.TenantID, input.CloudSessionID, nextRevision, input.ExpectedHeadRevision, + input.Bundle.BundleRef, input.Bundle.TransportDigest, input.Bundle.PayloadDigest, + input.Bundle.SizeBytes, input.ActivationID, + ); err != nil { + return storeport.CommitActivationResult{}, mapError("insert bundle revision", err) + } + + session.Status.HeadRevision = nextRevision + session.Status.LastCommittedActivationID = input.ActivationID + session.Status.ActiveActivationID = "" + session.Status.Phase = resource.CloudSessionReady + session.Status.ObservedGeneration = session.Metadata.Generation + if _, err := r.updateCloudSession(ctx, session); err != nil { + return storeport.CommitActivationResult{}, err + } + + activation.Status.Phase = resource.ActivationPublishing + activation.Status.CommittedRevision = &nextRevision + activation.Status.Outcome = &input.Outcome + if _, err := r.updateActivation(ctx, activation); err != nil { + return storeport.CommitActivationResult{}, err + } + + var outboxID int64 + if err := r.q.QueryRow(ctx, ` + INSERT INTO publication_outbox( + tenant_id,cloud_session_id,activation_id,destination,payload + ) VALUES($1,$2,$3,$4,$5) + RETURNING outbox_id`, + input.TenantID, input.CloudSessionID, input.ActivationID, + input.Publication.Destination, input.Publication.Payload, + ).Scan(&outboxID); err != nil { + return storeport.CommitActivationResult{}, mapError("enqueue publication", err) + } + tag, err := r.q.Exec(ctx, ` + UPDATE session_leases + SET resource_version=resource_version+1, + renewed_at=clock_timestamp(),expires_at=clock_timestamp() + WHERE tenant_id=$1 AND cloud_session_id=$2 AND epoch=$3 + AND holder_id=$4 AND activation_id=$5 + AND expires_at>clock_timestamp()`, + input.TenantID, input.CloudSessionID, input.LeaseEpoch, + input.HolderID, input.ActivationID, + ) + if err != nil { + return storeport.CommitActivationResult{}, mapError("expire committed lease", err) + } + if tag.RowsAffected() != 1 { + return storeport.CommitActivationResult{}, conflict("commit activation lease") + } + if err := r.appendChange(ctx, input.TenantID, storeport.ResourceKindBundleRevision, bundleRevisionResourceID(input.CloudSessionID, nextRevision), strconv.FormatInt(nextRevision, 10)); err != nil { + return storeport.CommitActivationResult{}, err + } + return storeport.CommitActivationResult{Revision: nextRevision, OutboxID: outboxID}, nil +} + +func (r *sessionTx) replayCommittedActivation(ctx context.Context, activation resource.Activation, input storeport.CommitActivationRequest) (storeport.CommitActivationResult, error) { + if activation.Status.Outcome == nil { + return storeport.CommitActivationResult{}, invalid("replay committed activation", "committed activation has no outcome") + } + if activation.Status.LeaseEpoch == nil || *activation.Status.LeaseEpoch != input.LeaseEpoch { + return storeport.CommitActivationResult{}, duplicate("replay committed activation", "lease epoch differs from committed activation") + } + var bundle resource.BundleRevision + if err := r.q.QueryRow(ctx, ` + SELECT tenant_id,cloud_session_id,revision,parent_revision,bundle_ref, + transport_digest,payload_digest,size_bytes,created_by_activation_id,created_at + FROM bundle_revisions + WHERE tenant_id=$1 AND cloud_session_id=$2 AND created_by_activation_id=$3`, + input.TenantID, input.CloudSessionID, input.ActivationID, + ).Scan( + &bundle.TenantID, &bundle.CloudSessionID, &bundle.Revision, &bundle.ParentRevision, + &bundle.BundleRef, &bundle.TransportDigest, &bundle.PayloadDigest, &bundle.SizeBytes, + &bundle.CreatedByActivationID, &bundle.CreatedAt, + ); err != nil { + return storeport.CommitActivationResult{}, mapError("read committed bundle", err) + } + var outboxID int64 + var destination string + var payload json.RawMessage + if err := r.q.QueryRow(ctx, ` + SELECT outbox_id,destination,payload + FROM publication_outbox + WHERE tenant_id=$1 AND cloud_session_id=$2 AND activation_id=$3`, + input.TenantID, input.CloudSessionID, input.ActivationID, + ).Scan(&outboxID, &destination, &payload); err != nil { + return storeport.CommitActivationResult{}, mapError("read committed publication", err) + } + storedOutcome, err := canonicalOutcome(*activation.Status.Outcome) + if err != nil { + return storeport.CommitActivationResult{}, err + } + requestOutcome, err := canonicalOutcome(input.Outcome) + if err != nil { + return storeport.CommitActivationResult{}, err + } + storedPayload, err := canonicalJSON(payload) + if err != nil { + return storeport.CommitActivationResult{}, err + } + requestPayload, err := canonicalJSON(input.Publication.Payload) + if err != nil { + return storeport.CommitActivationResult{}, err + } + if bundle.ParentRevision != input.ExpectedHeadRevision || + bundle.BundleRef != input.Bundle.BundleRef || + bundle.TransportDigest != input.Bundle.TransportDigest || + bundle.PayloadDigest != input.Bundle.PayloadDigest || + bundle.SizeBytes != input.Bundle.SizeBytes || + destination != input.Publication.Destination || + !bytes.Equal(storedOutcome, requestOutcome) || + !bytes.Equal(storedPayload, requestPayload) { + return storeport.CommitActivationResult{}, duplicate("replay committed activation", "commit input differs from committed result") + } + return storeport.CommitActivationResult{Revision: bundle.Revision, OutboxID: outboxID}, nil +} + +func (r *sessionTx) updateCloudSession(ctx context.Context, input resource.CloudSession) (resource.CloudSession, error) { + expected, err := parseVersion(input.Metadata.ResourceVersion, "update cloud session") + if err != nil { + return resource.CloudSession{}, err + } + input.Metadata.ResourceVersion = strconv.FormatInt(expected+1, 10) + metadata, spec, status, err := marshalCloudSession(input) + if err != nil { + return resource.CloudSession{}, err + } + result, err := scanCloudSession(r.q.QueryRow(ctx, ` + UPDATE cloud_sessions + SET resource_version=$4,metadata=$5,spec=$6,status=$7,lifecycle=$8, + phase=$9,head_revision=$10,active_activation_id=NULLIF($11,''),updated_at=clock_timestamp() + WHERE tenant_id=$1 AND cloud_session_id=$2 AND resource_version=$3 + RETURNING metadata,spec,status`, + input.Metadata.TenantID, input.Metadata.ID, expected, expected+1, + metadata, spec, status, input.Spec.Lifecycle, input.Status.Phase, + input.Status.HeadRevision, input.Status.ActiveActivationID, + ), "update cloud session") + if err != nil { + return resource.CloudSession{}, err + } + if err := r.recordChange(ctx, result.Metadata.TenantID, storeport.ResourceKindCloudSession, result.Metadata.ID, result.Metadata.ResourceVersion); err != nil { + return resource.CloudSession{}, err + } + return result, nil +} + +func (r *sessionTx) updateActivation(ctx context.Context, input resource.Activation) (resource.Activation, error) { + expected, err := parseVersion(input.Metadata.ResourceVersion, "update activation") + if err != nil { + return resource.Activation{}, err + } + input.Metadata.ResourceVersion = strconv.FormatInt(expected+1, 10) + metadata, spec, status, err := marshalActivation(input) + if err != nil { + return resource.Activation{}, err + } + var outcome []byte + if input.Status.Outcome != nil { + outcome, err = json.Marshal(input.Status.Outcome) + if err != nil { + return resource.Activation{}, invalidErr("marshal activation outcome", err) + } + } + result, _, err := scanActivation(r.q.QueryRow(ctx, ` + UPDATE activations + SET resource_version=$5,metadata=$6,spec=$7,status=$8,phase=$9, + base_revision=$10,lease_epoch=$11,committed_revision=$12, + outcome=$13::jsonb,updated_at=clock_timestamp() + WHERE tenant_id=$1 AND cloud_session_id=$2 AND activation_id=$3 + AND resource_version=$4 + RETURNING metadata,spec,status,request_hash`, + input.Metadata.TenantID, input.Spec.CloudSessionID, input.Spec.ActivationID, + expected, expected+1, metadata, spec, status, input.Status.Phase, + input.Status.BaseRevision, input.Status.LeaseEpoch, input.Status.CommittedRevision, + outcome, + ), "update activation") + if err != nil { + return resource.Activation{}, err + } + if err := r.recordChange(ctx, result.Metadata.TenantID, storeport.ResourceKindActivation, result.Metadata.ID, result.Metadata.ResourceVersion); err != nil { + return resource.Activation{}, err + } + return result, nil +} + +func (r *sessionTx) recordChange(ctx context.Context, tenantID, kind, id, version string) error { + r.state.mutations[mutationKey{tenant: tenantID, kind: kind, id: id, version: version}] = true + return r.appendChange(ctx, tenantID, kind, id, version) +} + +func (r *sessionTx) appendChange(ctx context.Context, tenantID, kind, id, version string) error { + _, err := (&changeRepo{q: r.q, state: r.state}).AppendResourceChange( + ctx, tenantID, kind, id, version, time.Now().UTC(), + ) + return err +} + +func getCloudSession(ctx context.Context, q querier, tenantID, sessionID string, lock bool) (resource.CloudSession, error) { + if !validIDs(tenantID, sessionID) { + return resource.CloudSession{}, invalid("get cloud session", "empty identity") + } + query := `SELECT metadata,spec,status FROM cloud_sessions WHERE tenant_id=$1 AND cloud_session_id=$2` + if lock { + query += ` FOR UPDATE` + } + result, err := scanCloudSession(q.QueryRow(ctx, query, tenantID, sessionID), "get cloud session") + if err == nil && (result.Metadata.TenantID != tenantID || result.Metadata.ID != sessionID) { + return resource.CloudSession{}, invalid("get cloud session", "persisted identity mismatch") + } + return result, err +} + +func getActivation(ctx context.Context, q querier, tenantID, sessionID, activationID string, lock bool) (resource.Activation, []byte, error) { + if !validIDs(tenantID, sessionID, activationID) { + return resource.Activation{}, nil, invalid("get activation", "empty identity") + } + query := ` + SELECT metadata,spec,status,request_hash + FROM activations + WHERE tenant_id=$1 AND cloud_session_id=$2 AND activation_id=$3` + if lock { + query += ` FOR UPDATE` + } + result, hash, err := scanActivation(q.QueryRow(ctx, query, tenantID, sessionID, activationID), "get activation") + if err == nil && (result.Metadata.TenantID != tenantID || result.Spec.CloudSessionID != sessionID || result.Spec.ActivationID != activationID) { + return resource.Activation{}, nil, invalid("get activation", "persisted identity mismatch") + } + return result, hash, err +} + +func getSessionLease(ctx context.Context, q querier, tenantID, sessionID string, lock bool) (resource.SessionLease, bool, error) { + query := ` + SELECT tenant_id,cloud_session_id,holder_id,activation_id,epoch, + resource_version::text,acquired_at,renewed_at,expires_at + FROM session_leases + WHERE tenant_id=$1 AND cloud_session_id=$2` + if lock { + query += ` FOR UPDATE` + } + var result resource.SessionLease + err := q.QueryRow(ctx, query, tenantID, sessionID).Scan( + &result.TenantID, &result.CloudSessionID, &result.HolderID, &result.ActivationID, + &result.Epoch, &result.ResourceVersion, &result.AcquiredAt, &result.RenewedAt, + &result.ExpiresAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return resource.SessionLease{}, false, nil + } + if err != nil { + return resource.SessionLease{}, false, mapError("get session lease", err) + } + result.AcquiredAt = result.AcquiredAt.UTC() + result.RenewedAt = result.RenewedAt.UTC() + result.ExpiresAt = result.ExpiresAt.UTC() + return result, true, nil +} + +func scanCloudSession(row pgx.Row, operation string) (resource.CloudSession, error) { + var metadata, spec, status []byte + if err := row.Scan(&metadata, &spec, &status); err != nil { + return resource.CloudSession{}, mapError(operation, err) + } + var result resource.CloudSession + if err := json.Unmarshal(metadata, &result.Metadata); err != nil { + return result, invalidErr(operation, err) + } + if err := json.Unmarshal(spec, &result.Spec); err != nil { + return result, invalidErr(operation, err) + } + if err := json.Unmarshal(status, &result.Status); err != nil { + return result, invalidErr(operation, err) + } + if !validIDs(result.Metadata.TenantID, result.Metadata.ID, result.Metadata.ResourceVersion) { + return result, invalid(operation, "invalid persisted cloud session metadata") + } + return result, nil +} + +func scanActivation(row pgx.Row, operation string) (resource.Activation, []byte, error) { + var metadata, spec, status, hash []byte + if err := row.Scan(&metadata, &spec, &status, &hash); err != nil { + return resource.Activation{}, nil, mapError(operation, err) + } + var result resource.Activation + if err := json.Unmarshal(metadata, &result.Metadata); err != nil { + return result, nil, invalidErr(operation, err) + } + if err := json.Unmarshal(spec, &result.Spec); err != nil { + return result, nil, invalidErr(operation, err) + } + if err := json.Unmarshal(status, &result.Status); err != nil { + return result, nil, invalidErr(operation, err) + } + if !validIDs(result.Metadata.TenantID, result.Metadata.ID, result.Metadata.ResourceVersion, result.Spec.CloudSessionID, result.Spec.ActivationID) { + return result, nil, invalid(operation, "invalid persisted activation metadata") + } + return result, hash, nil +} + +func marshalCloudSession(input resource.CloudSession) ([]byte, []byte, []byte, error) { + metadata, err := json.Marshal(input.Metadata) + if err != nil { + return nil, nil, nil, invalidErr("marshal cloud session metadata", err) + } + spec, err := json.Marshal(input.Spec) + if err != nil { + return nil, nil, nil, invalidErr("marshal cloud session spec", err) + } + status, err := json.Marshal(input.Status) + if err != nil { + return nil, nil, nil, invalidErr("marshal cloud session status", err) + } + return metadata, spec, status, nil +} + +func marshalActivation(input resource.Activation) ([]byte, []byte, []byte, error) { + metadata, err := json.Marshal(input.Metadata) + if err != nil { + return nil, nil, nil, invalidErr("marshal activation metadata", err) + } + spec, err := json.Marshal(input.Spec) + if err != nil { + return nil, nil, nil, invalidErr("marshal activation spec", err) + } + status, err := json.Marshal(input.Status) + if err != nil { + return nil, nil, nil, invalidErr("marshal activation status", err) + } + return metadata, spec, status, nil +} + +func normalizeNewCloudSession(input resource.CloudSession) (resource.CloudSession, error) { + if !validIDs(input.Metadata.TenantID, input.Metadata.ID, input.Spec.AgentRef) || input.Spec.AgentGeneration <= 0 { + return resource.CloudSession{}, invalid("create cloud session", "tenant, session, agent, and positive agent generation are required") + } + if input.Metadata.ResourceVersion != "" || input.Metadata.Generation != 0 || + input.Metadata.DeletionTimestamp != nil || len(input.Metadata.Finalizers) != 0 { + return resource.CloudSession{}, invalid("create cloud session", "server-owned metadata must be empty") + } + if input.Status != (resource.CloudSessionStatus{}) { + return resource.CloudSession{}, invalid("create cloud session", "server-owned initial status must be empty") + } + if input.Spec.Lifecycle == "" { + input.Spec.Lifecycle = resource.CloudSessionLifecycleActive + } + if input.Spec.Lifecycle != resource.CloudSessionLifecycleActive && input.Spec.Lifecycle != resource.CloudSessionLifecycleSuspended { + return resource.CloudSession{}, invalid("create cloud session", "invalid lifecycle") + } + input.Metadata.Generation = 1 + input.Metadata.ResourceVersion = "1" + input.Metadata.Finalizers = []string{} + input.Status.ObservedGeneration = 1 + if input.Spec.Lifecycle == resource.CloudSessionLifecycleSuspended { + input.Status.Phase = resource.CloudSessionSuspended + } else { + input.Status.Phase = resource.CloudSessionReady + } + return input, nil +} + +func normalizeNewActivation(input resource.Activation) (resource.Activation, error) { + if !validIDs(input.Metadata.TenantID, input.Spec.CloudSessionID, input.Spec.ActivationID) { + return resource.Activation{}, invalid("admit activation", "tenant, session, and activation identity are required") + } + if input.Metadata.ID == "" { + input.Metadata.ID = input.Spec.ActivationID + } + if input.Metadata.ResourceVersion != "" || input.Metadata.Generation != 0 || + input.Metadata.DeletionTimestamp != nil || len(input.Metadata.Finalizers) != 0 || + input.Spec.AgentGeneration != 0 { + return resource.Activation{}, invalid("admit activation", "server-owned metadata and agent generation must be empty") + } + if !zeroActivationStatus(input.Status) { + return resource.Activation{}, invalid("admit activation", "server-owned initial status must be empty") + } + switch input.Spec.Stimulus.Type { + case resource.ActivationStimulusMessage, resource.ActivationStimulusSchedule, resource.ActivationStimulusSystem: + default: + return resource.Activation{}, invalid("admit activation", "invalid stimulus type") + } + payload, err := canonicalJSON(input.Spec.Stimulus.Payload) + if err != nil { + return resource.Activation{}, invalidErr("admit activation payload", err) + } + input.Spec.Stimulus.Payload = payload + if input.Spec.Deadline != nil { + value := input.Spec.Deadline.UTC() + input.Spec.Deadline = &value + } + input.Metadata.Generation = 1 + input.Metadata.ResourceVersion = "1" + input.Metadata.Finalizers = []string{} + input.Status.ObservedGeneration = 1 + input.Status.Phase = resource.ActivationPending + return input, nil +} + +func normalizeCommitRequest(input storeport.CommitActivationRequest) (storeport.CommitActivationRequest, error) { + if !validIDs(input.TenantID, input.CloudSessionID, input.ActivationID, input.HolderID, + input.Bundle.BundleRef, input.Bundle.TransportDigest, input.Bundle.PayloadDigest, + input.Publication.Destination) || + input.LeaseEpoch <= 0 || input.ExpectedHeadRevision < 0 || input.Bundle.SizeBytes < 0 { + return input, invalid("commit activation", "invalid identity, lease, head, bundle, or publication") + } + switch input.Outcome.Status { + case resource.ActivationOutcomeCompleted, resource.ActivationOutcomeBlocked, + resource.ActivationOutcomeRetryableFailure, resource.ActivationOutcomeFatalFailure: + default: + return input, invalid("commit activation", "invalid outcome status") + } + if len(input.Outcome.Response) != 0 { + value, err := canonicalJSON(input.Outcome.Response) + if err != nil { + return input, invalidErr("commit activation outcome", err) + } + input.Outcome.Response = value + } + payload, err := canonicalJSON(input.Publication.Payload) + if err != nil { + return input, invalidErr("commit activation publication", err) + } + input.Publication.Payload = payload + return input, nil +} + +func activationRequestHash(spec resource.ActivationSpec) ([]byte, error) { + payload, err := canonicalJSON(spec.Stimulus.Payload) + if err != nil { + return nil, invalidErr("hash activation request", err) + } + value := struct { + CloudSessionID string + ActivationID string + AgentGeneration int64 + StimulusType resource.ActivationStimulusType + StimulusPayload json.RawMessage + Deadline *time.Time + }{ + CloudSessionID: spec.CloudSessionID, ActivationID: spec.ActivationID, + AgentGeneration: spec.AgentGeneration, StimulusType: spec.Stimulus.Type, + StimulusPayload: payload, Deadline: spec.Deadline, + } + encoded, err := json.Marshal(value) + if err != nil { + return nil, invalidErr("hash activation request", err) + } + sum := sha256.Sum256(encoded) + return sum[:], nil +} + +func canonicalOutcome(input resource.ActivationOutcome) ([]byte, error) { + if len(input.Response) != 0 { + value, err := canonicalJSON(input.Response) + if err != nil { + return nil, invalidErr("canonicalize activation outcome", err) + } + input.Response = value + } + encoded, err := json.Marshal(input) + if err != nil { + return nil, invalidErr("canonicalize activation outcome", err) + } + return encoded, nil +} + +// canonicalJSON sorts object keys through encoding/json while UseNumber keeps +// integers beyond float64 precision exact. +func canonicalJSON(input []byte) ([]byte, error) { + if len(bytes.TrimSpace(input)) == 0 { + return nil, errors.New("empty JSON") + } + decoder := json.NewDecoder(bytes.NewReader(input)) + decoder.UseNumber() + value, err := decodeJSONValue(decoder) + if err != nil { + return nil, err + } + if _, err := decoder.Token(); err == nil { + return nil, errors.New("multiple JSON values") + } else if !errors.Is(err, io.EOF) { + return nil, err + } + return json.Marshal(value) +} + +func decodeJSONValue(decoder *json.Decoder) (any, error) { + token, err := decoder.Token() + if err != nil { + return nil, err + } + delim, composite := token.(json.Delim) + if !composite { + return token, nil + } + switch delim { + case '{': + object := map[string]any{} + for decoder.More() { + keyToken, err := decoder.Token() + if err != nil { + return nil, err + } + key, ok := keyToken.(string) + if !ok { + return nil, errors.New("object key is not a string") + } + if _, duplicate := object[key]; duplicate { + return nil, fmt.Errorf("duplicate object key %q", key) + } + value, err := decodeJSONValue(decoder) + if err != nil { + return nil, err + } + object[key] = value + } + if token, err := decoder.Token(); err != nil || token != json.Delim('}') { + if err != nil { + return nil, err + } + return nil, errors.New("unterminated object") + } + return object, nil + case '[': + array := []any{} + for decoder.More() { + value, err := decodeJSONValue(decoder) + if err != nil { + return nil, err + } + array = append(array, value) + } + if token, err := decoder.Token(); err != nil || token != json.Delim(']') { + if err != nil { + return nil, err + } + return nil, errors.New("unterminated array") + } + return array, nil + default: + return nil, fmt.Errorf("unexpected JSON delimiter %q", delim) + } +} + +func databaseNow(ctx context.Context, q querier) (time.Time, error) { + var value time.Time + if err := q.QueryRow(ctx, `SELECT clock_timestamp()`).Scan(&value); err != nil { + return time.Time{}, mapError("read database time", err) + } + return value.UTC(), nil +} + +func parseVersion(value, operation string) (int64, error) { + version, err := strconv.ParseInt(value, 10, 64) + if err != nil || version <= 0 { + return 0, invalid(operation, "invalid resource version") + } + return version, nil +} + +func zeroActivationStatus(value resource.ActivationStatus) bool { + return value.ObservedGeneration == 0 && value.Phase == "" && value.Attempt == 0 && + value.BaseRevision == nil && value.SandboxAttemptID == "" && value.LeaseEpoch == nil && + value.CommittedRevision == nil && value.Outcome == nil +} + +func duplicate(operation, message string) error { + return &storeport.Error{Kind: storeport.ErrorDuplicate, Operation: operation, Err: errors.New(message)} +} + +func bundleRevisionResourceID(sessionID string, revision int64) string { + return strconv.Itoa(len(sessionID)) + ":" + sessionID + "/" + strconv.FormatInt(revision, 10) +} diff --git a/internal/store/postgres/session_integration_test.go b/internal/store/postgres/session_integration_test.go new file mode 100644 index 0000000..1f298ff --- /dev/null +++ b/internal/store/postgres/session_integration_test.go @@ -0,0 +1,451 @@ +package postgres + +import ( + "context" + "encoding/json" + "fmt" + "os" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/maka-agent/maka-agent-cloud/internal/resource" + "github.com/maka-agent/maka-agent-cloud/internal/store" +) + +func TestPostgresSessionCommitProtocol(t *testing.T) { + url := os.Getenv("MAKA_POSTGRES_TEST_URL") + if url == "" { + t.Skip("set MAKA_POSTGRES_TEST_URL to run disposable-schema PostgreSQL tests") + } + ctx := context.Background() + pool, cfg := sessionTestPool(t, ctx, url) + if err := ApplyMigration(ctx, pool, MigrationUp); err != nil { + t.Fatal(err) + } + if err := ApplyMigration(ctx, pool, MigrationUp); err != nil { + t.Fatalf("idempotent migration: %v", err) + } + authority, err := New(pool) + if err != nil { + t.Fatal(err) + } + sessions := authority.Sessions() + + inputSession := resource.CloudSession{ + Metadata: resource.Metadata{TenantID: "tenant-a", ID: "session-a"}, + Spec: resource.CloudSessionSpec{ + AgentRef: "agent-a", AgentGeneration: 7, + }, + } + createdSession, err := sessions.CreateCloudSession(ctx, inputSession) + if err != nil { + t.Fatal(err) + } + if createdSession.Metadata.ResourceVersion != "1" || + createdSession.Status.Phase != resource.CloudSessionReady || + createdSession.Status.HeadRevision != 0 { + t.Fatalf("created session=%+v", createdSession) + } + if _, err := sessions.GetCloudSession(ctx, "tenant-b", "session-a"); !store.IsKind(err, store.ErrorNotFound) { + t.Fatalf("session tenant isolation: %v", err) + } + otherTenantSession := inputSession + otherTenantSession.Metadata.TenantID = "tenant-b" + if _, err := sessions.CreateCloudSession(ctx, otherTenantSession); err != nil { + t.Fatalf("same session ID in another tenant: %v", err) + } + otherTenantActivation := activationInput("tenant-b", "session-a", "activation-1", `{"text":"other tenant"}`) + if _, created, err := sessions.AdmitActivation(ctx, otherTenantActivation); err != nil || !created { + t.Fatalf("same activation ID in another tenant created=%v err=%v", created, err) + } + + firstInput := activationInput("tenant-a", "session-a", "activation-1", `{"text":"first","n":9007199254740993}`) + secondInput := activationInput("tenant-a", "session-a", "activation-2", `{"text":"second"}`) + first, firstCreated, err := sessions.AdmitActivation(ctx, firstInput) + if err != nil || !firstCreated { + t.Fatalf("admit first created=%v err=%v", firstCreated, err) + } + if first.Spec.AgentGeneration != 7 || first.Status.Phase != resource.ActivationPending { + t.Fatalf("admitted first=%+v", first) + } + if _, created, err := sessions.AdmitActivation(ctx, firstInput); err != nil || created { + t.Fatalf("idempotent admission created=%v err=%v", created, err) + } + different := firstInput + different.Spec.Stimulus.Payload = json.RawMessage(`{"n":9007199254740992,"text":"first"}`) + if _, _, err := sessions.AdmitActivation(ctx, different); !store.IsKind(err, store.ErrorDuplicate) { + t.Fatalf("large-integer idempotency conflict=%v", err) + } + if _, created, err := sessions.AdmitActivation(ctx, secondInput); err != nil || !created { + t.Fatalf("admit second created=%v err=%v", created, err) + } + + if _, _, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-a", "activation-2", "worker-2", time.Minute); !store.IsKind(err, store.ErrorConflict) { + t.Fatalf("FIFO violation accepted: %v", err) + } + lease, acquired, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-a", "activation-1", "worker-1", time.Minute) + if err != nil || !acquired || lease.Epoch != 1 { + t.Fatalf("acquire first lease=%+v acquired=%v err=%v", lease, acquired, err) + } + replayedLease, acquired, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-a", "activation-1", "worker-1", time.Minute) + if err != nil || acquired || replayedLease.ResourceVersion != lease.ResourceVersion { + t.Fatalf("lease replay=%+v acquired=%v err=%v", replayedLease, acquired, err) + } + if _, _, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-a", "activation-1", "other-worker", time.Minute); !store.IsKind(err, store.ErrorConflict) { + t.Fatalf("live lease stolen: %v", err) + } + lease, err = sessions.RenewSessionLease(ctx, lease, time.Minute) + if err != nil || lease.ResourceVersion != "2" { + t.Fatalf("renew lease=%+v err=%v", lease, err) + } + + firstCommit := store.CommitActivationRequest{ + TenantID: "tenant-a", + CloudSessionID: "session-a", + ActivationID: "activation-1", + HolderID: "worker-1", + LeaseEpoch: lease.Epoch, + ExpectedHeadRevision: 0, + Bundle: store.BundleCandidate{ + BundleRef: "s3://bundles/session-a/1", TransportDigest: "sha256:transport-1", + PayloadDigest: "sha256:payload-1", SizeBytes: 101, + }, + Outcome: resource.ActivationOutcome{ + Status: resource.ActivationOutcomeCompleted, + Response: json.RawMessage( + `{"usage":{"tokens":9007199254740993},"text":"done"}`, + ), + }, + Publication: store.Publication{ + Destination: "conversation:1", + Payload: json.RawMessage(`{"text":"done","sequence":9007199254740993}`), + }, + } + results := make(chan store.CommitActivationResult, 2) + errors := make(chan error, 2) + var group sync.WaitGroup + for range 2 { + group.Add(1) + go func() { + defer group.Done() + result, err := sessions.CommitActivation(ctx, firstCommit) + results <- result + errors <- err + }() + } + group.Wait() + close(results) + close(errors) + for err := range errors { + if err != nil { + t.Fatalf("concurrent idempotent commit: %v", err) + } + } + var committed store.CommitActivationResult + for result := range results { + if committed == (store.CommitActivationResult{}) { + committed = result + continue + } + if result != committed { + t.Fatalf("commit results differ: first=%+v second=%+v", committed, result) + } + } + if committed.Revision != 1 || committed.OutboxID <= 0 { + t.Fatalf("commit result=%+v", committed) + } + conflictingCommit := firstCommit + conflictingCommit.Publication.Payload = json.RawMessage(`{"sequence":9007199254740992,"text":"done"}`) + if _, err := sessions.CommitActivation(ctx, conflictingCommit); !store.IsKind(err, store.ErrorDuplicate) { + t.Fatalf("changed commit replay accepted: %v", err) + } + + afterFirst, err := sessions.GetCloudSession(ctx, "tenant-a", "session-a") + if err != nil { + t.Fatal(err) + } + if afterFirst.Status.HeadRevision != 1 || + afterFirst.Status.LastCommittedActivationID != "activation-1" || + afterFirst.Status.ActiveActivationID != "" || + afterFirst.Status.Phase != resource.CloudSessionReady { + t.Fatalf("session after commit=%+v", afterFirst) + } + bundles, err := sessions.ListBundleRevisions(ctx, "tenant-a", "session-a") + if err != nil || len(bundles) != 1 || bundles[0].Revision != 1 { + t.Fatalf("bundles=%+v err=%v", bundles, err) + } + publications, err := sessions.ListPublicationOutbox(ctx, "tenant-a", 10) + if err != nil || len(publications) != 1 || publications[0].ID != committed.OutboxID { + t.Fatalf("publications=%+v err=%v", publications, err) + } + + secondLease, acquired, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-a", "activation-2", "worker-2", time.Minute) + if err != nil || !acquired || secondLease.Epoch <= lease.Epoch { + t.Fatalf("acquire second lease=%+v acquired=%v err=%v", secondLease, acquired, err) + } + secondCommit := store.CommitActivationRequest{ + TenantID: "tenant-a", CloudSessionID: "session-a", ActivationID: "activation-2", + HolderID: "worker-2", LeaseEpoch: secondLease.Epoch, ExpectedHeadRevision: 1, + Bundle: store.BundleCandidate{ + BundleRef: "s3://bundles/session-a/2", TransportDigest: "sha256:transport-2", + PayloadDigest: "sha256:payload-2", SizeBytes: 202, + }, + Outcome: resource.ActivationOutcome{ + Status: resource.ActivationOutcomeCompleted, + Response: json.RawMessage(`{"text":"second done"}`), + }, + Publication: store.Publication{ + Destination: "conversation:1", Payload: json.RawMessage(`{"text":"second done"}`), + }, + } + if _, err := sessions.CommitActivation(ctx, secondCommit); err != nil { + t.Fatal(err) + } + if replayed, err := sessions.CommitActivation(ctx, firstCommit); err != nil || replayed != committed { + t.Fatalf("old commit replay after later head=%+v err=%v", replayed, err) + } + staleWorker := secondCommit + staleWorker.HolderID = "worker-1" + staleWorker.LeaseEpoch = lease.Epoch + staleWorker.Bundle.BundleRef = "s3://bundles/session-a/stale" + if _, err := sessions.CommitActivation(ctx, staleWorker); !store.IsKind(err, store.ErrorDuplicate) { + t.Fatalf("stale worker changed committed turn: %v", err) + } + + changes, err := authority.ResourceChanges().ListResourceChanges(ctx, "tenant-a", "", 100) + if err != nil { + t.Fatal(err) + } + if len(changes) < 10 { + t.Fatalf("missing session resource changes: %+v", changes) + } + seenKinds := map[string]bool{} + for index, change := range changes { + seenKinds[change.ResourceKind] = true + if index > 0 { + before, _ := parseVersion(changes[index-1].ChangeID, "test change cursor") + current, _ := parseVersion(change.ChangeID, "test change cursor") + if current <= before { + t.Fatalf("resource changes out of order: %+v", changes) + } + } + } + for _, kind := range []string{ + store.ResourceKindCloudSession, + store.ResourceKindActivation, + store.ResourceKindBundleRevision, + } { + if !seenKinds[kind] { + t.Fatalf("resource changes missing kind %s: %+v", kind, changes) + } + } + + recoverySession := inputSession + recoverySession.Metadata.ID = "session-recovery" + if _, err := sessions.CreateCloudSession(ctx, recoverySession); err != nil { + t.Fatal(err) + } + recoveryActivation := activationInput("tenant-a", "session-recovery", "activation-recovery", `{"text":"recover"}`) + if _, _, err := sessions.AdmitActivation(ctx, recoveryActivation); err != nil { + t.Fatal(err) + } + oldLease, acquired, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-recovery", "activation-recovery", "old-worker", time.Minute) + if err != nil || !acquired { + t.Fatalf("acquire recovery lease=%+v acquired=%v err=%v", oldLease, acquired, err) + } + if _, err := pool.Exec(ctx, ` + UPDATE session_leases SET expires_at=clock_timestamp()-interval '1 second' + WHERE tenant_id='tenant-a' AND cloud_session_id='session-recovery'`); err != nil { + t.Fatal(err) + } + recoveredLease, acquired, err := sessions.AcquireSessionLease(ctx, "tenant-a", "session-recovery", "activation-recovery", "new-worker", time.Minute) + if err != nil || !acquired || recoveredLease.Epoch <= oldLease.Epoch { + t.Fatalf("expired lease recovery=%+v acquired=%v err=%v", recoveredLease, acquired, err) + } + if _, err := sessions.RenewSessionLease(ctx, oldLease, time.Minute); !store.IsKind(err, store.ErrorConflict) { + t.Fatalf("stale lease renewed: %v", err) + } + if _, err := pool.Exec(ctx, ` + UPDATE session_leases SET expires_at=clock_timestamp()-interval '1 second' + WHERE tenant_id='tenant-a' AND cloud_session_id='session-recovery'`); err != nil { + t.Fatal(err) + } + expiredCommit := store.CommitActivationRequest{ + TenantID: "tenant-a", CloudSessionID: "session-recovery", ActivationID: "activation-recovery", + HolderID: "new-worker", LeaseEpoch: recoveredLease.Epoch, ExpectedHeadRevision: 0, + Bundle: store.BundleCandidate{ + BundleRef: "s3://bundles/session-recovery/1", TransportDigest: "sha256:transport-r", + PayloadDigest: "sha256:payload-r", SizeBytes: 1, + }, + Outcome: resource.ActivationOutcome{ + Status: resource.ActivationOutcomeCompleted, Response: json.RawMessage(`{"text":"late"}`), + }, + Publication: store.Publication{ + Destination: "conversation:r", Payload: json.RawMessage(`{"text":"late"}`), + }, + } + if _, err := sessions.CommitActivation(ctx, expiredCommit); !store.IsKind(err, store.ErrorConflict) { + t.Fatalf("expired lease committed: %v", err) + } + recoveryBundles, err := sessions.ListBundleRevisions(ctx, "tenant-a", "session-recovery") + if err != nil || len(recoveryBundles) != 0 { + t.Fatalf("expired commit leaked bundle=%+v err=%v", recoveryBundles, err) + } + + pool.Close() + restartedPool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer restartedPool.Close() + restarted, err := New(restartedPool) + if err != nil { + t.Fatal(err) + } + recovered, err := restarted.Sessions().GetCloudSession(ctx, "tenant-a", "session-a") + if err != nil || recovered.Status.HeadRevision != 2 { + t.Fatalf("restart recovery=%+v err=%v", recovered, err) + } +} + +func TestPostgresMigrationBootstrapsLegacyBaseline(t *testing.T) { + url := os.Getenv("MAKA_POSTGRES_TEST_URL") + if url == "" { + t.Skip("set MAKA_POSTGRES_TEST_URL to run disposable-schema PostgreSQL tests") + } + ctx := context.Background() + pool, _ := sessionTestPool(t, ctx, url) + baseline, err := migrations.ReadFile("migrations/000001_sandbox_reconcile.up.sql") + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, string(baseline)); err != nil { + t.Fatal(err) + } + if err := ApplyMigration(ctx, pool, MigrationUp); err != nil { + t.Fatalf("upgrade legacy baseline: %v", err) + } + var versions []int64 + rows, err := pool.Query(ctx, `SELECT version FROM maka_cloud_schema_migrations ORDER BY version`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var version int64 + if err := rows.Scan(&version); err != nil { + t.Fatal(err) + } + versions = append(versions, version) + } + rows.Close() + if fmt.Sprint(versions) != "[1 2]" { + t.Fatalf("migration versions=%v", versions) + } + var sessionTable bool + if err := pool.QueryRow(ctx, `SELECT to_regclass('cloud_sessions') IS NOT NULL`).Scan(&sessionTable); err != nil { + t.Fatal(err) + } + if !sessionTable { + t.Fatal("session migration was not applied") + } +} + +func TestPostgresConcurrentFreshMigration(t *testing.T) { + url := os.Getenv("MAKA_POSTGRES_TEST_URL") + if url == "" { + t.Skip("set MAKA_POSTGRES_TEST_URL to run disposable-schema PostgreSQL tests") + } + ctx := context.Background() + pool, _ := sessionTestPool(t, ctx, url) + var group sync.WaitGroup + errors := make(chan error, 8) + for range 8 { + group.Add(1) + go func() { + defer group.Done() + errors <- ApplyMigration(ctx, pool, MigrationUp) + }() + } + group.Wait() + close(errors) + for err := range errors { + if err != nil { + t.Fatalf("concurrent fresh migration: %v", err) + } + } + var count int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM maka_cloud_schema_migrations`).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 2 { + t.Fatalf("migration count=%d want=2", count) + } +} + +func TestPostgresMigrationRoundTrip(t *testing.T) { + url := os.Getenv("MAKA_POSTGRES_TEST_URL") + if url == "" { + t.Skip("set MAKA_POSTGRES_TEST_URL to run disposable-schema PostgreSQL tests") + } + ctx := context.Background() + pool, _ := sessionTestPool(t, ctx, url) + if err := ApplyMigration(ctx, pool, MigrationUp); err != nil { + t.Fatal(err) + } + if err := ApplyMigration(ctx, pool, MigrationDown); err != nil { + t.Fatal(err) + } + var remaining int + if err := pool.QueryRow(ctx, ` + SELECT count(*) + FROM pg_catalog.pg_tables + WHERE schemaname=current_schema() + `).Scan(&remaining); err != nil { + t.Fatal(err) + } + if remaining != 0 { + t.Fatalf("tables remain after down migration: %d", remaining) + } +} + +func activationInput(tenantID, sessionID, activationID, payload string) resource.Activation { + return resource.Activation{ + Metadata: resource.Metadata{TenantID: tenantID}, + Spec: resource.ActivationSpec{ + CloudSessionID: sessionID, + ActivationID: activationID, + Stimulus: resource.ActivationStimulus{ + Type: resource.ActivationStimulusMessage, Payload: json.RawMessage(payload), + }, + }, + } +} + +func sessionTestPool(t *testing.T, ctx context.Context, url string) (*pgxpool.Pool, *pgxpool.Config) { + t.Helper() + admin, err := pgxpool.New(ctx, url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(admin.Close) + schema := fmt.Sprintf("maka_session_%d", time.Now().UnixNano()) + if _, err := admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _, _ = admin.Exec(context.Background(), "DROP SCHEMA "+schema+" CASCADE") + }) + cfg, err := pgxpool.ParseConfig(url) + if err != nil { + t.Fatal(err) + } + cfg.ConnConfig.RuntimeParams["search_path"] = schema + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + return pool, cfg +} diff --git a/internal/store/postgres/session_test.go b/internal/store/postgres/session_test.go new file mode 100644 index 0000000..f57e7de --- /dev/null +++ b/internal/store/postgres/session_test.go @@ -0,0 +1,122 @@ +package postgres + +import ( + "bytes" + "encoding/json" + "testing" + + "github.com/maka-agent/maka-agent-cloud/internal/resource" + "github.com/maka-agent/maka-agent-cloud/internal/store" +) + +func TestCanonicalJSONPreservesLargeIntegers(t *testing.T) { + left, err := canonicalJSON([]byte(`{"z":9007199254740993,"a":1}`)) + if err != nil { + t.Fatal(err) + } + right, err := canonicalJSON([]byte(`{"a":1,"z":9007199254740992}`)) + if err != nil { + t.Fatal(err) + } + if bytes.Equal(left, right) { + t.Fatalf("distinct integers collapsed: left=%s right=%s", left, right) + } + if got, want := string(left), `{"a":1,"z":9007199254740993}`; got != want { + t.Fatalf("canonical JSON=%s want=%s", got, want) + } + if empty, err := canonicalJSON([]byte(`[]`)); err != nil || string(empty) != `[]` { + t.Fatalf("empty array canonicalized as %s: %v", empty, err) + } + for _, invalid := range [][]byte{ + nil, + []byte(`{`), + []byte(`1 2`), + []byte(`{"a":1,"a":2}`), + []byte(`{"nested":{"a":1,"a":2}}`), + } { + if _, err := canonicalJSON(invalid); err == nil { + t.Fatalf("invalid JSON accepted: %q", invalid) + } + } +} + +func TestActivationRequestHashIsSemanticAndPrecise(t *testing.T) { + base := resource.ActivationSpec{ + CloudSessionID: "session", + ActivationID: "activation", + AgentGeneration: 3, + Stimulus: resource.ActivationStimulus{ + Type: resource.ActivationStimulusMessage, + Payload: json.RawMessage(`{"b":9007199254740993,"a":1}`), + }, + } + equivalent := base + equivalent.Stimulus.Payload = json.RawMessage(`{"a":1,"b":9007199254740993}`) + first, err := activationRequestHash(base) + if err != nil { + t.Fatal(err) + } + second, err := activationRequestHash(equivalent) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(first, second) { + t.Fatal("object key order changed activation identity") + } + different := base + different.Stimulus.Payload = json.RawMessage(`{"a":1,"b":9007199254740992}`) + third, err := activationRequestHash(different) + if err != nil { + t.Fatal(err) + } + if bytes.Equal(first, third) { + t.Fatal("large integer difference was lost") + } +} + +func TestServerOwnsInitialSessionAndActivationState(t *testing.T) { + session := resource.CloudSession{ + Metadata: resource.Metadata{TenantID: "tenant", ID: "session"}, + Spec: resource.CloudSessionSpec{ + AgentRef: "agent", AgentGeneration: 2, + }, + } + normalized, err := normalizeNewCloudSession(session) + if err != nil { + t.Fatal(err) + } + if normalized.Metadata.ResourceVersion != "1" || + normalized.Metadata.Generation != 1 || + normalized.Status.Phase != resource.CloudSessionReady || + normalized.Status.ObservedGeneration != 1 { + t.Fatalf("unexpected normalized session: %+v", normalized) + } + session.Status.Phase = resource.CloudSessionBusy + if _, err := normalizeNewCloudSession(session); !store.IsKind(err, store.ErrorInvalid) { + t.Fatalf("caller-owned session status accepted: %v", err) + } + + activation := resource.Activation{ + Metadata: resource.Metadata{TenantID: "tenant"}, + Spec: resource.ActivationSpec{ + CloudSessionID: "session", + ActivationID: "activation", + Stimulus: resource.ActivationStimulus{ + Type: resource.ActivationStimulusMessage, Payload: json.RawMessage(`{"text":"hello"}`), + }, + }, + } + admitted, err := normalizeNewActivation(activation) + if err != nil { + t.Fatal(err) + } + if admitted.Metadata.ID != "activation" || + admitted.Metadata.ResourceVersion != "1" || + admitted.Status.Phase != resource.ActivationPending { + t.Fatalf("unexpected normalized activation: %+v", admitted) + } + activation.Status.Phase = resource.ActivationRunning + if _, err := normalizeNewActivation(activation); !store.IsKind(err, store.ErrorInvalid) { + t.Fatalf("caller-owned activation status accepted: %v", err) + } +} diff --git a/internal/store/postgres/store.go b/internal/store/postgres/store.go index bfd6c91..c31dfd9 100644 --- a/internal/store/postgres/store.go +++ b/internal/store/postgres/store.go @@ -52,14 +52,19 @@ func (s *Store) Close() { func (s *Store) SandboxAttempts() store.SandboxAttemptRepository { return &attemptRepo{q: s.pool} } func (s *Store) ResourceChanges() store.ResourceChangeRepository { return &changeRepo{q: s.pool} } func (s *Store) ReconcileOutbox() store.ReconcileOutboxRepository { return &outboxRepo{pool: s.pool} } +func (s *Store) Sessions() store.SessionRepository { return &sessionStore{authority: s} } -type mutationKey struct{ tenant, id, version string } +type mutationKey struct{ tenant, kind, id, version string } type txState struct{ mutations, changes map[mutationKey]bool } func (s *Store) Transact(ctx context.Context, fn func(store.Transaction) error) error { if fn == nil { return invalid("transact", "nil callback") } + return s.transact(ctx, func(tx *transaction) error { return fn(tx) }) +} + +func (s *Store) transact(ctx context.Context, fn func(*transaction) error) error { tx, err := s.pool.Begin(ctx) if err != nil { return mapError("begin transaction", err) @@ -195,7 +200,7 @@ func (r *attemptRepo) FinalizeSandboxAttempt(ctx context.Context, tenant, id, ve return got, nil } func (r *attemptRepo) mark(a resource.SandboxAttempt) { - r.state.mutations[mutationKey{a.Metadata.TenantID, a.Metadata.ID, a.Metadata.ResourceVersion}] = true + r.state.mutations[mutationKey{a.Metadata.TenantID, store.ResourceKindSandboxAttempt, a.Metadata.ID, a.Metadata.ResourceVersion}] = true } func (r *attemptRepo) cas(ctx context.Context, t, id string, err error, op string) error { if !store.IsKind(err, store.ErrorNotFound) { @@ -220,15 +225,16 @@ func (r *changeRepo) AppendResourceChange(ctx context.Context, t, k, id, v strin if !validIDs(t, k, id, v) || at.IsZero() { return store.ResourceChange{}, invalid("append resource change", "invalid input") } + if _, e := r.q.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, t); e != nil { + return store.ResourceChange{}, mapError("lock tenant resource changes", e) + } var n int64 var ts time.Time e := r.q.QueryRow(ctx, `INSERT INTO resource_changes(tenant_id,resource_kind,resource_id,resource_version,occurred_at) VALUES($1,$2,$3,$4,$5) RETURNING change_id,occurred_at`, t, k, id, v, at).Scan(&n, &ts) if e != nil { return store.ResourceChange{}, mapError("append resource change", e) } - if k == store.ResourceKindSandboxAttempt { - r.state.changes[mutationKey{t, id, v}] = true - } + r.state.changes[mutationKey{t, k, id, v}] = true return store.ResourceChange{ChangeID: strconv.FormatInt(n, 10), TenantID: t, ResourceKind: k, ResourceID: id, ResourceVersion: v, OccurredAt: ts.UTC()}, nil } func (r *changeRepo) ListResourceChanges(ctx context.Context, t, after string, limit int) ([]store.ResourceChange, error) { diff --git a/internal/store/postgres/store_test.go b/internal/store/postgres/store_test.go index cb6a20d..1ed8b52 100644 --- a/internal/store/postgres/store_test.go +++ b/internal/store/postgres/store_test.go @@ -15,7 +15,17 @@ func TestMigrationContract(t *testing.T) { if err != nil { t.Fatal(err) } - for _, want := range []string{"CREATE TABLE sandbox_attempts", "CREATE TABLE resource_changes", "CREATE TABLE reconcile_outbox", "FOR UPDATE"} { + for _, want := range []string{ + "CREATE TABLE sandbox_attempts", + "CREATE TABLE resource_changes", + "CREATE TABLE reconcile_outbox", + "CREATE TABLE cloud_sessions", + "CREATE TABLE activations", + "CREATE TABLE session_leases", + "CREATE TABLE bundle_revisions", + "CREATE TABLE publication_outbox", + "FOR UPDATE", + } { if want == "FOR UPDATE" { continue } @@ -36,6 +46,9 @@ func TestMigrationContract(t *testing.T) { "metadata->>'ID' = sandbox_attempt_id", "metadata->>'ResourceVersion' = resource_version::text", "metadata->>'Generation')::bigint = generation", + "spec->>'Lifecycle' = lifecycle", + "status->>'HeadRevision')::bigint = head_revision", + "spec->>'ActivationID' = activation_id", } { if !strings.Contains(up, invariant) { t.Errorf("up migration missing metadata invariant %q", invariant) @@ -45,11 +58,21 @@ func TestMigrationContract(t *testing.T) { if err != nil { t.Fatal(err) } - for _, table := range []string{"reconcile_outbox", "resource_changes", "sandbox_attempts"} { + for _, table := range []string{ + "publication_outbox", "bundle_revisions", "session_leases", + "activations", "cloud_sessions", "reconcile_outbox", + "resource_changes", "sandbox_attempts", + } { if !strings.Contains(down, "DROP TABLE IF EXISTS "+table) { t.Errorf("down migration does not drop %s", table) } } + if strings.Index(up, "000001_sandbox_reconcile.up.sql") > strings.Index(up, "000002_session_commit_protocol.up.sql") { + t.Error("up migrations are not ordered by version") + } + if strings.Index(down, "000002_session_commit_protocol.down.sql") > strings.Index(down, "000001_sandbox_reconcile.down.sql") { + t.Error("down migrations are not reverse ordered") + } } func TestErrorMappingDoesNotLeakPGX(t *testing.T) { diff --git a/internal/store/session.go b/internal/store/session.go new file mode 100644 index 0000000..3461479 --- /dev/null +++ b/internal/store/session.go @@ -0,0 +1,72 @@ +package store + +import ( + "context" + "encoding/json" + "time" + + "github.com/maka-agent/maka-agent-cloud/internal/resource" +) + +const ( + ResourceKindCloudSession = "CloudSession" + ResourceKindActivation = "Activation" + ResourceKindBundleRevision = "BundleRevision" +) + +type BundleCandidate struct { + BundleRef string + TransportDigest string + PayloadDigest string + SizeBytes int64 +} + +type Publication struct { + Destination string + Payload json.RawMessage +} + +type CommitActivationRequest struct { + TenantID string + CloudSessionID string + ActivationID string + HolderID string + LeaseEpoch int64 + ExpectedHeadRevision int64 + Bundle BundleCandidate + Outcome resource.ActivationOutcome + Publication Publication +} + +type CommitActivationResult struct { + Revision int64 + OutboxID int64 +} + +type PublicationOutboxEntry struct { + ID int64 + TenantID string + CloudSessionID string + ActivationID string + Destination string + Payload json.RawMessage + Attempts int + AvailableAt time.Time + CreatedAt time.Time + DeliveredAt *time.Time +} + +// SessionRepository owns the PostgreSQL consistency boundary for one +// CloudSession turn. Mutations include their resource-change records and +// publication intent in the same authoritative transaction. +type SessionRepository interface { + CreateCloudSession(context.Context, resource.CloudSession) (resource.CloudSession, error) + GetCloudSession(context.Context, string, string) (resource.CloudSession, error) + AdmitActivation(context.Context, resource.Activation) (resource.Activation, bool, error) + GetActivation(context.Context, string, string, string) (resource.Activation, error) + AcquireSessionLease(context.Context, string, string, string, string, time.Duration) (resource.SessionLease, bool, error) + RenewSessionLease(context.Context, resource.SessionLease, time.Duration) (resource.SessionLease, error) + CommitActivation(context.Context, CommitActivationRequest) (CommitActivationResult, error) + ListBundleRevisions(context.Context, string, string) ([]resource.BundleRevision, error) + ListPublicationOutbox(context.Context, string, int) ([]PublicationOutboxEntry, error) +}