Build every advisory key from its family, and take locks only through the helpers

Second of two changes for #186; closes it. Each kind of object the
console serializes on now has a family, and every advisory key is built
one way: hashtextextended('<family>:<id>', 0), with a UUID written in
canonical form. db.Lock and db.LockShared (internal/db/locks.go) build
the key and take the lock on a *sql.Tx only, and each family's
statement begins with a name line, so pg_stat_activity and the test
traces show which family a backend waits on.

Families, in the order a transaction takes them: materialization,
domains-registry, fedwiki-workspace, lifecycle, stripe-subscription,
stripe-invoice, stripe-product-sync, desired-state. The lifecycle and
desired-state keys are unchanged byte for byte; the other six change.
Migration 00040 replaces core.assert_rendezvous to look for the new
materialization key (owner, grants, SECURITY DEFINER and search_path
kept; Down restores migration 18's body), and both seeds take the new
key right after BEGIN.

docs/database-locks.md lists the families in a table and writes one
order every path keeps: the materialization lock first, then the
families in table order, then rows. The project lint gains
advisory-lock-site (an advisory lock function named in a Go string or
in SQL or shell code outside the helpers, the test kit's probe, the
migrations that name the rendezvous and the rule's own tests; a seed
taking any other key, or the rendezvous anywhere but first after
BEGIN), pool-lock-site (a core.resource_pools row lock outside
LockResourcePool and LockGrantPool, or a call or method value of those
outside entitlements/tx.go) and lock-families (the doc's table against
the code). The walk covers cmd/, scripts/, internal/, test/ and the
root's Go files, and skips testdata and directories whose name begins
with a dot, as the go tool does. Tests take a family's lock through the
same helpers and probe one with dbtest.TryLock.

Recording a lifecycle request comes in two forms:
RecordLifecycleRequest opens its own transaction on a database or a
pinned connection, and RecordLifecycleRequestInTx runs in the caller's
*sql.Tx, so a connection is never used outside a transaction (which
released the instance lock at once) and does not compile in the second.
The same holds for RecordFirstLifecycleRequest.

docs/production-deployment.md now states the upgrade order for every
upgrade: stop every process of the running release, run migrate up if
migrations run separately, then start the new release; two releases
running at once can build different keys for one lock and share none.
docs/go-conventions.md, docs/testing.md and
docs/building-an-integration.md follow the code. The lint command's
help text names the files the lock rules read. 202 golden lines in 109
files change their statement text.

Deploy: stop every process of the previous release first. Rollback:
stop this release; as the migration user, in one transaction, run
00040's Down section and DELETE FROM goose_db_version_core WHERE
version_id = 40; start the previous release (migrate down rolls back
the last integration stream first). A deployment using console_reader
re-runs scripts/db-reader-role.sql.

Break list: 35 records, 35 CAUGHT. Named without a record: the *sql.Tx
parameters of the helpers and the -InTx forms (a break that passes a
connection does not compile, and the build is the check), and the
comments. Full suite on reset databases: 5727 passed, 16 skipped, none
failed.
This commit is contained in:
2026-10-31 04:03:32 -05:00
parent 0ee4924b9d
commit e52fdb37c8
178 changed files with 3520 additions and 531 deletions
+11 -4
View File
@@ -15,10 +15,11 @@ import (
var lintCmd = &cobra.Command{
Use: "lint",
Short: "Project lint: routes, swap targets, page anatomy and copy",
Long: `lint walks the Go source under internal/, the templates under
internal/embeds/templates/, the scripts under internal/embeds/static/ and the
walkthroughs and screens manifest under test/e2e/ and the specs under
openspec/specs/, and reports:
Long: `lint walks the Go source under internal/ (and, for the lock rules, the Go,
SQL and shell files under cmd/, scripts/, internal/ and test/ and the root's
Go files), the templates under internal/embeds/templates/, the scripts under
internal/embeds/static/ and the walkthroughs and screens manifest under
test/e2e/ and the specs under openspec/specs/, and reports:
- routes: routeURL patterns with no registered handler
- swap targets: hx-target="#X" with no matching id="X"
@@ -34,6 +35,9 @@ openspec/specs/, and reports:
- spec purposes: a spec whose Purpose is the archive's placeholder or empty
- templates: a template that does not parse in its surface, or that no
surface parses
- lock sites: an advisory or pool row lock taken outside its helper,
a seed's other key, or a family table that disagrees
with internal/db
Exits 0 on a clean tree, 1 on violations, 2 on internal errors. See
docs/testing.md, "Static checks".`,
@@ -45,6 +49,7 @@ docs/testing.md, "Static checks".`,
allowlistPath, _ := cmd.Flags().GetString("anatomy-allowlist")
manifestPath, _ := cmd.Flags().GetString("screens-manifest")
specsDir, _ := cmd.Flags().GetString("specs-dir")
lockRoot, _ := cmd.Flags().GetString("lock-root")
res, err := lint.Run(lint.Config{
ServerDir: serverDir,
@@ -53,6 +58,7 @@ docs/testing.md, "Static checks".`,
AllowlistPath: allowlistPath,
ManifestPath: manifestPath,
SpecsDir: specsDir,
LockRoot: lockRoot,
})
if err != nil {
fmt.Fprintf(os.Stderr, "lint: %v\n", err)
@@ -77,5 +83,6 @@ func init() {
lintCmd.Flags().String("screens-manifest", "test/e2e/screens/manifest.go", `Screens manifest for the screen-coverage rule ("-" disables it)`)
lintCmd.Flags().String("walkthrough-dir", "test/e2e/operator-walkthroughs", `Walkthrough directory for coverage rule (set "-" to disable)`)
lintCmd.Flags().String("specs-dir", "openspec/specs", `Specs directory for the spec-purpose-placeholder rule (set "-" to disable)`)
lintCmd.Flags().String("lock-root", ".", `Repository root the lock rules walk ("-" disables the lock rules)`)
rootCmd.AddCommand(lintCmd)
}
+1 -1
View File
@@ -589,7 +589,7 @@ Use this decision rule when designing a new integration's mutating verbs:
FarmManager and write `fedwiki.sites` in the same workflow) — **dispatch
directly** via your own `RegisterWorkflows`-registered workflows. When a
person starts the operation, record it first as a lifecycle request
through `internal/integration`'s writers (`RecordLifecycleRequest` in the
through `internal/integration`'s writers (`RecordLifecycleRequestInTx` in the
same transaction as your own rows, then start the workflow under the
request's workflow id, and finish the request from the workflow), as
FedWiki does: the handler then answers when the workflow starts, and a
+38 -35
View File
@@ -8,9 +8,7 @@ summary: "Every advisory lock and serializing row lock the console takes, how ea
This page describes the locks as the code takes them today. It records what
each lock serializes, who takes it, and the order each path takes several.
It does not describe a target design. One piece of planned work changes the
page: #186 gives every advisory key one family prefix and writes one lock
order. Whoever lands it edits this page in the same change.
It does not describe a target design.
The code names functions and sqlc queries; so does this page. Line numbers
are left out because they move.
@@ -27,14 +25,14 @@ creators lock exactly one site `FOR SHARE` (see Registry below).
| Lock | Key | Mode | Taken by | Serializes |
|---|---|---|---|---|
| Materialization rendezvous | `hashtextextended('entitlement_materialization', 0)` | shared for a transaction that materializes a pool; exclusive for a rule change, for a change to a ladder's tiers (tier add, tier delete, tier removal, the reorder's renumbering) and for a default change's T0 | the `entitlements.Materializing` and `entitlements.RuleChange` transaction kinds (`BeginMaterializing`, `BeginRuleChange`), as the transaction's first statement | A rule change or a tier change against every materializing transaction: each one finishes before the change or starts after it, so a conferral reads one set of rules and one order of each ladder. Materializing transactions do not wait on each other here. |
| Subscription | `hashtext(<Stripe subscription id>)::bigint` | exclusive | `fulfillment.ReconcileSubscription` | Two reconciles of one subscription (the checkout return and a webhook), so they converge instead of racing the exclusion constraint on subscription periods. |
| Invoice | `hashtext(<Stripe invoice id>)::bigint` | exclusive | `invoiceReconcile.converge` in `internal/integrations/stripe/workflows` | Two reconciles of one invoice (two events, or an event and the sweep), so the second finds what the first wrote. |
| Product sync | `hashtext(<product id>)::bigint` | exclusive | `catalogsync.Sync` in `internal/integrations/stripe/catalogsync` | Two sync clicks for one product, so the mapping writes and the outbox enqueue happen once. |
| FedWiki workspace | `hashtext(<workspace id>)::bigint` | exclusive | `withWorkspaceTx` in `internal/integrations/fedwiki/web`, for the status-change and swap handlers | One workspace's rotations of its active site, so two tabs cannot interleave the cooldown check and stamp or the choice of the site to park. |
| Lifecycle instance | `hashtextextended('lifecycle:' \|\| <instance id>, 0)` | exclusive | the `LockLifecycleInstance` query, run by `integration.RecordLifecycleRequest`, `RecordFirstLifecycleRequest` and the FedWiki activity helpers `finishAllWith` and `purgeArchivedSite` | One instance's requests: the latest-generation check and the insert are one step, and an outcome is recorded against a request that is still accepted; and retention's check that no request is open, held through its farm delete and local deletion. |
| Desired-state address | `hashtextextended('desired-state:' \|\| <connection id> \|\| ':' \|\| <recipient kind> \|\| ':' \|\| <recipient id>, 0)` | exclusive | the `LockDesiredStateAddress` query, run by `lockAddress` in `internal/workflows/desiredstate` | Every write to the content or generation of the record at one (connection, recipient) address, and to its projection, including an address that has no record yet. |
| Domains registry | the constant `registryAllocationLock` (the ASCII bytes of "domains") | exclusive | `domains.Registry.WithLock` | Every mutating allocation in the registry, one lock for the whole namespace, because the disjointness rules span rows no database constraint can compare. |
| Materialization rendezvous | `hashtextextended('materialization:', 0)` | shared for a transaction that materializes a pool; exclusive for a rule change, for a change to a ladder's tiers (tier add, tier delete, tier removal, the reorder's renumbering) and for a default change's T0 | the `entitlements.Materializing` and `entitlements.RuleChange` transaction kinds (`BeginMaterializing`, `BeginRuleChange`), as the transaction's first statement; `test/mockshot-seed.sql` and `test/seed/discourse/seed-discourse.sh`, as their first statement after `BEGIN` | A rule change or a tier change against every materializing transaction: each one finishes before the change or starts after it, so a conferral reads one set of rules and one order of each ladder. Materializing transactions do not wait on each other here. |
| Subscription | `hashtextextended('stripe-subscription:<Stripe subscription id>', 0)` | exclusive | `fulfillment.ReconcileSubscription` | Two reconciles of one subscription (the checkout return and a webhook), so they converge instead of racing the exclusion constraint on subscription periods. |
| Invoice | `hashtextextended('stripe-invoice:<Stripe invoice id>', 0)` | exclusive | `invoiceReconcile.converge` in `internal/integrations/stripe/workflows` | Two reconciles of one invoice (two events, or an event and the sweep), so the second finds what the first wrote. |
| Product sync | `hashtextextended('stripe-product-sync:<product id>', 0)` | exclusive | `catalogsync.Sync` in `internal/integrations/stripe/catalogsync` | Two sync clicks for one product, so the mapping writes and the outbox enqueue happen once. |
| FedWiki workspace | `hashtextextended('fedwiki-workspace:<workspace id>', 0)` | exclusive | `withWorkspaceTx` in `internal/integrations/fedwiki/web`, for the status-change and swap handlers | One workspace's rotations of its active site, so two tabs cannot interleave the cooldown check and stamp or the choice of the site to park. |
| Lifecycle instance | `hashtextextended('lifecycle:<instance id>', 0)` | exclusive | `db.Lock` on the lifecycle family, run by `integration.RecordLifecycleRequestInTx` and `RecordFirstLifecycleRequestInTx`, FedWiki's `RecordSiteRequestInTx` and `CheckSwapRequestSites`, `usage.Hold`, and the activity helpers `finishAllWith` and `purgeArchivedSite` | One instance's requests: the latest-generation check and the insert are one step, and an outcome is recorded against a request that is still accepted; and retention's check that no request is open, held through its farm delete and local deletion. |
| Desired-state address | `hashtextextended('desired-state:<connection id>:<recipient kind>:<recipient id>', 0)` | exclusive | `lockAddress` in `internal/workflows/desiredstate` | Every write to the content or generation of the record at one (connection, recipient) address, and to its projection, including an address that has no record yet. |
| Domains registry | `hashtextextended('domains-registry:', 0)` | exclusive | `registryKind`, the kind of `domains.Registry.WithLock` | Every mutating allocation in the registry, one lock for the whole namespace, because the disjointness rules span rows no database constraint can compare. |
### Row locks
@@ -60,25 +58,30 @@ flag-only writes change neither claims nor allocation counts.
| Desired-state notice | `core.desired_state_outbox` rows | `FOR UPDATE SKIP LOCKED` | the `ClaimDesiredStateOutbox` query | Two relay passes, so each claims different rows; a pass skips rows another still holds. Triggers write the rows through `core.insert_desired_state_notice` inside the transaction that changed the source, and they take no lock beyond the insert. |
| Resource-key row | `core.resource_keys` row of one key | the update lock of `SetResourceKeyProvider` | `integration.RegisterProviders` at boot, stamping keys in byte order | A stamp that changes a key's provider against an open entitlement insert of that key, which holds the row's `FOR KEY SHARE` through the foreign key. The materializer writes a pool's keys in byte order too, so the two wait in one direction. |
## How advisory keys are built
## Advisory key families
Postgres takes one 64-bit key for an advisory lock, so every lock above maps
its identity onto a bigint. The code builds those keys four ways.
Every advisory key is `hashtextextended('<family>:<id>', 0)`: the family's name, a colon and the id, whose parts are joined by colons. `db.Lock` and `db.LockShared` (`internal/db/locks.go`) build it and take it, on a transaction, and nothing else names an advisory lock function (the project lint's `advisory-lock-site`). A UUID part is written in canonical form, as `uuid::text` writes it, and a text part holds no colon; a family with one lock for the whole database has no id. Each family's statement begins with a name line (`-- name: LockSubscription :exec`), which `pg_stat_activity` shows for a waiter and the statement traces record. The table lists the families in the order a transaction takes them; the lint fails when it and `db.Families` differ (`lock-families`).
1. **`hashtextextended` of a fixed name.** The rendezvous. One key for the
whole database.
2. **`hashtextextended` of a prefixed string.** The lifecycle instance
(`lifecycle:` and the instance id) and the desired-state address
(`desired-state:`, the connection, the recipient kind and the recipient
id). The prefix keeps the two families from sharing keys with each other.
3. **`hashtext` of a bare id, widened to bigint.** The subscription, the
invoice, the product sync and the FedWiki workspace. `hashtext` returns 32
bits, so these four families share one 32-bit key space with no prefix
between them; a collision serializes two unrelated objects briefly and
cannot lose a write. The ids differ in shape (Stripe ids against UUIDs),
which makes a cross-family collision unlikely, not impossible.
4. **A literal constant.** The domains registry. Its value is larger than any
32-bit hash, so it cannot collide with family 3.
| Family | Id | Statement | Taken by |
|---|---|---|---|
| `materialization` | none | `LockMaterialization`, `LockMaterializationShared` | the `entitlements.Materializing` and `entitlements.RuleChange` kinds; `test/mockshot-seed.sql` and `test/seed/discourse/seed-discourse.sh`, as their first statement; `core.assert_rendezvous` checks its key |
| `domains-registry` | none | `LockDomainsRegistry` | `registryKind`, the kind of `domains.Registry.WithLock` |
| `fedwiki-workspace` | workspace id | `LockFedWikiWorkspace` | `withWorkspaceTx` |
| `lifecycle` | instance id | `LockLifecycleInstance` | the lifecycle recording, FedWiki's intake, the slot gate, the outcome steps and the retention purge |
| `stripe-subscription` | Stripe subscription id | `LockSubscription` | `fulfillment.ReconcileSubscription`, `supersedeReleasedDowngrades` |
| `stripe-invoice` | Stripe invoice id | `LockInvoice` | `invoiceReconcile.converge` |
| `stripe-product-sync` | product id | `LockProductSync` | `catalogsync.Sync` |
| `desired-state` | connection id, recipient kind, recipient id | `LockDesiredStateAddress` | `lockAddress` in `internal/workflows/desiredstate` |
A change to a family's name or to the shape of its id changes its keys. Two processes that build different keys for one object hold no common lock, so a release that changes a key is deployed as [production-deployment.md](production-deployment.md), section 5, orders an upgrade.
## The order every path keeps
1. A transaction that takes the materialization rendezvous takes it as its first statement. No other advisory lock comes before it.
2. The advisory families the transaction needs, in the order of the table above. Several locks of one family follow the order this page gives for that family: lifecycle instances in the order their requests were recorded.
3. Rows, by the rules below: a rule change's set and rule rows before pool rows; pool rows in ascending `pool_id`, before a grant's lineage head, obligation rows and a tier's product row; FedWiki's site rows, then their placement rows through the trigger, then the usage row, then request rows; a desired-state record after its address.
No path waits on an advisory lock after it takes a row lock: a recording that takes an instance lock its transaction already holds, as the swap's second recording does, takes it without waiting. The sections below list each path's locks in this order.
## The order each path takes locks
@@ -539,9 +542,11 @@ writes no count; the insert, the status write and the delete commit.
generation, then inserts. `RecordFirstLifecycleRequest` takes the same lock
once and reads the latest generation once: it refuses a later generation of an
instance that already holds a request, then decides as
`RecordLifecycleRequest` does. Called with a `*sql.DB`, each opens its own
transaction; called with a transaction, each uses the caller's, which is how
the handler records under the workspace lock.
`RecordLifecycleRequest` does. `RecordLifecycleRequest` and
`RecordFirstLifecycleRequest` open their own transaction on a database or a
pinned connection; `RecordLifecycleRequestInTx` and
`RecordFirstLifecycleRequestInTx` run in the caller's, which is how the handler
records under the workspace lock.
`DeleteUnstartedLifecycleRequest`, which removes a request whose workflow never
started, takes no instance lock, so a repeat that races it may be answered from
either side of the delete.
@@ -586,7 +591,7 @@ registry operations; skipped sites do not incur this wait.
`registryKind`, whose first statement is the registry lock. Every composite
operation, including the FedWiki create, release and sync projections, runs on
the `*Tx` it passes, so the lock is taken once. The comment on
`registryAllocationLock` states the rule for a transaction that would touch both
`registryKind` states the rule for a transaction that would touch both
the registry and a usage row: registry lock first. No transaction does both
today: the slot gate and the recounts are workflow activities that run their own
transactions, apart from the transaction that allocates or releases the name.
@@ -603,7 +608,7 @@ transactions, apart from the transaction that allocates or releases the name.
adds its ladder's tiers on the plain handle, outside the rendezvous.
`core.assert_rendezvous` raises
`materialization_rendezvous_missing` when a function that materializes or
settles runs without it.
settles runs without it. No other advisory lock precedes it.
2. The pool row is locked before positions or entitlements change. The five
conferral and settlement functions lock the row themselves, so a caller
that goes straight to one of them still holds it.
@@ -642,5 +647,3 @@ transactions, apart from the transaction that allocates or releases the name.
inside `core.end_conferral`, and takes it again for resumption and before
the floor restoration. Each path then runs resumption, the floor
restoration and materialization in the same order.
- **Advisory keys are built four ways** (#186). The keys follow the four
conventions above.
+14 -6
View File
@@ -146,12 +146,20 @@ are those two calls). A failed begin, a failed commit and `fn`'s own error
come back unwrapped; the caller adds its own context once around the whole
act.
A helper that takes a lock and then runs more statements under it, its own
or its caller's, takes a `*sql.Tx` rather than sqlc's `DBTX` (`usage.Recount`,
`usage.Hold`, and FedWiki's intake helpers `RecordSiteRequestInTx` and
`CheckSwapRequestSites`). Every advisory lock is transaction-scoped, so given
a `*sql.DB` each statement runs in a transaction of its own, the lock ends
with the statement that took it, and nothing reports the loss.
A helper that takes a lock and then runs more statements under it, its own or
its caller's, takes a `*sql.Tx` rather than sqlc's `DBTX`: an advisory lock
(`usage.Hold`, `integration.RecordLifecycleRequestInTx`, and FedWiki's intake
helpers `RecordSiteRequestInTx` and `CheckSwapRequestSites`) or a row lock
(`usage.Recount`, the usage row). Both kinds last until the transaction ends,
so given a `*sql.DB` each statement runs in a transaction of its own, the lock
ends with the statement that took it, and nothing reports the loss. An
advisory lock goes through `db.Lock` or `db.LockShared`
(`internal/db/locks.go`) with its family and id; the project lint refuses an
advisory lock function anywhere else, and a pool row lock outside `LockPools`
and `LockPoolOfGrant`. Those two are the one exception to the `*sql.Tx` rule:
they stay methods of `*entitlements.Queries`, because the plan act runner's
querier seam wraps them in tests and every caller builds `Queries` on a
transaction.
No call to Stripe runs while a transaction is open. A path reads from Stripe
before it opens its transaction, and writes to Stripe after the commit or
+17 -6
View File
@@ -160,12 +160,13 @@ also uses `db-migrate-dsn`.
**Run migrations as a separate step.** To keep the migration password out of
the running process, leave `db-migrate-dsn` off the service and run
`member-console migrate up` with it set as a one-shot step before each
deploy. With `db-migrate-dsn` unset, `start` runs no migration, and it
refuses to start while the schema is behind the code. After the first run of
`migrate up`, run, as the migration user, the `GRANT console_runtime TO
<runtime user> WITH INHERIT TRUE, SET FALSE;` statement that the refusal names, because
only a boot with `db-migrate-dsn` set grants it.
`member-console migrate up` with it set as a one-shot step after the old
release stops and before the new one starts (below). With `db-migrate-dsn`
unset, `start` runs no migration, and it refuses to start while the schema is
behind the code. After the first run of `migrate up`, run, as the migration
user, the `GRANT console_runtime TO <runtime user> WITH INHERIT TRUE, SET
FALSE;` statement that the refusal names, because only a boot with
`db-migrate-dsn` set grants it.
**Check configuration before restarting.** `member-console config validate`
connects to the database, parses every stored override and the resolved
@@ -176,6 +177,16 @@ upgrade's restart so a bad value (an environment variable edited by hand, or
an override saved before this check existed) surfaces in your terminal
instead of the service manager's restart loop.
**Stop the old release before starting the new one.** Stop every process of
the running release, run `member-console migrate up` if you run migrations as
a separate step, then start the new release. Two releases running at once can
build different keys for one advisory lock and then hold no common lock, so
writes the lock serializes can interleave
([database-locks.md](database-locks.md#advisory-key-families)); and a
migration can change what the old release's writes need. An orchestrator's
rolling update that starts the new task before it stops the old one, or a
service with more than one replica, is stopped or scaled to zero first.
**Manage stored overrides from the terminal.** `member-console config`
offers the operator panel's three override actions from the command line:
`config list [integration]` prints each declared key's environment or
+6 -1
View File
@@ -136,6 +136,10 @@ that unit tests can't reach.
- `dbtest.AwaitLockWait` polls until a backend waits on a lock (`Waiters` for a
number of backends queued at once), for a test that holds one transaction
open while a second one blocks.
- `dbtest.TryLock` probes whether a family's lock is free, under the key the
code builds. A test that holds a family's lock takes it with `db.Lock`, and
one that waits on it gives `dbtest.Backend` the family's `Statement()`, so
the test meets the key and the statement the code uses.
**Fixtures.** `world` (`internal/testkit/world`) builds the rows a test needs
through the functions the console itself calls. Its builders cover people,
@@ -313,6 +317,7 @@ checks, and `make lint-templates` is a sixth that runs on its own:
- copy: `em-dash-copy`, `filler-copy`;
- specs: `spec-purpose-placeholder`, a spec under `openspec/specs/` whose
Purpose is the archive's placeholder or empty;
- locks: `advisory-lock-site`, `pool-lock-site`, `lock-families`;
- htmx: `htmx-2-compat`, `htmx-old-event`, `hx-disinherit`;
- error text: `raw-error-render`;
- reading: `unreadable-file`, a directory the lint cannot list or a file it
@@ -491,7 +496,7 @@ a test of its own.
| Package | Holds | Imports |
|---|---|---|
| `testkit` | `Close`, which closes an `io.Closer` and fails the test on its error | no console package |
| `testkit/dbtest` | opening the test databases, scratch databases, counting the statements a request runs, `Q`, `AwaitLockWait` | `internal/db` and `internal/testkit` only (`internal/migrate` reaches nearly every package) |
| `testkit/dbtest` | opening the test databases, scratch databases, counting the statements a request runs, `Q`, `AwaitLockWait`, `TryLock` | `internal/db` and `internal/testkit` only (`internal/migrate` reaches nearly every package) |
| `testkit/htmltest` | parsing HTML and selecting from it with CSS | no console package |
| `testkit/golden` | comparing a test's output with a file under testdata, `-update` writing it, and naming the ids and times a run makes | no console package |
| `testkit/replay` | replaying a package's recorded Temporal histories against its workflows, and recording them | no console package |
+2 -2
View File
@@ -31,8 +31,8 @@ const (
// refusal, materialization_rendezvous_missing.
notInPrerequisiteState = "55000"
exclusiveRendezvous = `SELECT pg_advisory_xact_lock(hashtextextended('entitlement_materialization', 0))`
sharedRendezvous = `SELECT pg_advisory_xact_lock_shared(hashtextextended('entitlement_materialization', 0))`
exclusiveRendezvous = `SELECT pg_advisory_xact_lock(hashtextextended('materialization:', 0))`
sharedRendezvous = `SELECT pg_advisory_xact_lock_shared(hashtextextended('materialization:', 0))`
// recordTierRemoved calls core.record_plan_act for a tier removal by a
// person: $1 the person, $2 the ladder, $3 the product.
+139
View File
@@ -0,0 +1,139 @@
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package db
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"github.com/google/uuid"
)
// idPart is the shape of one part of a family's id.
type idPart int
const (
// textPart is used as given and may be neither empty nor hold ':', which
// would run into the next part: a Stripe id, a kind.
textPart idPart = iota
// uuidPart is parsed and written in canonical form, as Postgres writes
// uuid::text, so two spellings of one id take one key.
uuidPart
)
// Family is one advisory lock family. Its key for an id is
// hashtextextended(name || ':' || id, 0), the id's parts joined by ':'; a
// family with one lock for the whole database has an id of no parts. Only
// this file declares one. docs/database-locks.md lists the families in the
// order a transaction takes them, and the project lint holds its table to
// Families.
type Family struct {
name string
label string
parts []idPart
}
var (
// Materialization is the rendezvous: every transaction that materializes
// a pool holds it shared, every rule change and operator tier change
// exclusive (entitlements.Materializing, entitlements.RuleChange).
// core.assert_rendezvous checks its key.
Materialization = Family{name: "materialization", label: "LockMaterialization"}
// DomainsRegistry serializes every mutating allocation of a name
// (domains.Registry.WithLock).
DomainsRegistry = Family{name: "domains-registry", label: "LockDomainsRegistry"}
// FedWikiWorkspace serializes one workspace's rotations of its active
// site.
FedWikiWorkspace = Family{name: "fedwiki-workspace", label: "LockFedWikiWorkspace", parts: []idPart{uuidPart}}
// Lifecycle serializes one instance's lifecycle requests.
Lifecycle = Family{name: "lifecycle", label: "LockLifecycleInstance", parts: []idPart{uuidPart}}
// Subscription serializes two reconciles of one Stripe subscription.
Subscription = Family{name: "stripe-subscription", label: "LockSubscription", parts: []idPart{textPart}}
// Invoice serializes two reconciles of one Stripe invoice.
Invoice = Family{name: "stripe-invoice", label: "LockInvoice", parts: []idPart{textPart}}
// ProductSync serializes two Stripe syncs of one product.
ProductSync = Family{name: "stripe-product-sync", label: "LockProductSync", parts: []idPart{uuidPart}}
// DesiredState serializes every write at one address: a connection, a
// recipient kind and a recipient.
DesiredState = Family{name: "desired-state", label: "LockDesiredStateAddress", parts: []idPart{uuidPart, textPart, uuidPart}}
)
// Families is every family, in the order a transaction takes them.
var Families = []Family{Materialization, DomainsRegistry, FedWikiWorkspace, Lifecycle, Subscription, Invoice, ProductSync, DesiredState}
// Name is the family's name, the prefix of its keys.
func (f Family) Name() string { return f.name }
// Key is the text whose hashtextextended is f's key for the id's parts. It
// refuses a family this file did not declare, a number of parts other than
// the family's, an empty text part, a text part that holds ':' and a uuid
// part that does not parse.
func (f Family) Key(id ...string) (string, error) {
if f.name == "" {
return "", errors.New("advisory lock: no family")
}
if len(id) != len(f.parts) {
return "", fmt.Errorf("advisory lock %s: %d id parts, want %d", f.name, len(id), len(f.parts))
}
parts := make([]string, len(id))
for i, p := range id {
switch f.parts[i] {
case uuidPart:
u, err := uuid.Parse(p)
if err != nil {
return "", fmt.Errorf("advisory lock %s: id part %d: %w", f.name, i+1, err)
}
parts[i] = u.String()
default:
if p == "" {
return "", fmt.Errorf("advisory lock %s: id part %d is empty", f.name, i+1)
}
if strings.Contains(p, ":") {
return "", fmt.Errorf("advisory lock %s: id part %d holds ':'", f.name, i+1)
}
parts[i] = p
}
}
return f.name + ":" + strings.Join(parts, ":"), nil
}
// Statement is the statement Lock runs for f, the key as $1. Its first line
// names the family, which pg_stat_activity shows for a waiter and the
// statement traces record; a test waits on it by its text.
func (f Family) Statement() string {
return "-- name: " + f.label + " :exec\nSELECT pg_advisory_xact_lock(hashtextextended($1, 0))"
}
// SharedStatement is the statement LockShared runs for f.
func (f Family) SharedStatement() string {
return "-- name: " + f.label + "Shared :exec\nSELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))"
}
// Lock takes f's exclusive transaction-scoped advisory lock on the id,
// released at commit or rollback. It takes a *sql.Tx because on a database
// or a connection outside a transaction the lock would end with the
// statement that took it. An id Key refuses sends no statement; the
// driver's error comes back as it is.
func Lock(ctx context.Context, tx *sql.Tx, f Family, id ...string) error {
return take(ctx, tx, f.Statement(), f, id)
}
// LockShared takes f's shared lock on the id, as Lock takes the exclusive
// one.
func LockShared(ctx context.Context, tx *sql.Tx, f Family, id ...string) error {
return take(ctx, tx, f.SharedStatement(), f, id)
}
// take builds f's key for id and runs stmt with it on tx.
func take(ctx context.Context, tx *sql.Tx, stmt string, f Family, id []string) error {
key, err := f.Key(id...)
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, stmt, key)
return err
}
+163
View File
@@ -0,0 +1,163 @@
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package db_test
import (
"context"
"database/sql"
"strings"
"testing"
"github.com/google/uuid"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
)
// lockRow is one family, an id it locks and another id of the same family;
// a family with one lock for the whole database has neither.
type lockRow struct {
family db.Family
id []string
other []string
}
// lockRows builds a row for each family from ids no other test locks.
func lockRows() []lockRow {
u := func() string { return uuid.NewString() }
return []lockRow{
{family: db.Materialization},
{family: db.DomainsRegistry},
{family: db.FedWikiWorkspace, id: []string{u()}, other: []string{u()}},
{family: db.Lifecycle, id: []string{u()}, other: []string{u()}},
{family: db.Subscription, id: []string{"sub_" + u()}, other: []string{"sub_" + u()}},
{family: db.Invoice, id: []string{"in_" + u()}, other: []string{"in_" + u()}},
{family: db.ProductSync, id: []string{u()}, other: []string{u()}},
{family: db.DesiredState, id: []string{u(), "person", u()}, other: []string{u(), "person", u()}},
}
}
// holderOf begins a transaction on d that the test rolls back when it ends
// and returns it with its backend.
func holderOf(t *testing.T, d *sql.DB) (*sql.Tx, int) {
t.Helper()
q := dbtest.Tx(t, d)
return q.Tx(), int(q.Int(`SELECT pg_backend_pid()`))
}
// TestLockTakesOneKeyPerId holds that a second transaction asking for a
// family's lock on an id waits on the first, that the statement it runs
// names the family, that the key granted is hashtextextended of the family's
// key text, and that another id of the family is free.
func TestLockTakesOneKeyPerId(t *testing.T) {
d := dbtest.Open(t)
for _, row := range lockRows() {
t.Run(row.family.Name(), func(t *testing.T) { checkLockTakesOneKey(t, d, row) })
}
}
// checkLockTakesOneKey is one row of TestLockTakesOneKeyPerId.
func checkLockTakesOneKey(t *testing.T, d *sql.DB, row lockRow) {
t.Helper()
ctx := context.Background()
name := row.family.Name()
key, err := row.family.Key(row.id...)
if err != nil {
t.Fatalf("%s: key: %v", name, err)
}
holder, holderPID := holderOf(t, d)
if err := db.Lock(ctx, holder, row.family, row.id...); err != nil {
t.Fatalf("%s: the holder's lock: %v", name, err)
}
waiterDone := make(chan error, 1)
go func() {
tx, err := d.BeginTx(ctx, nil)
if err != nil {
waiterDone <- err
return
}
defer tx.Rollback()
waiterDone <- db.Lock(ctx, tx, row.family, row.id...)
}()
dbtest.AwaitLockWait(t, d, dbtest.Backend{BlockedBy: holderPID})
var query string
if err := d.QueryRowContext(ctx, `SELECT query FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid))`, holderPID).Scan(&query); err != nil {
t.Fatalf("%s: read the waiter's statement: %v", name, err)
}
first, _, _ := strings.Cut(row.family.Statement(), "\n")
if !strings.HasPrefix(query, first) {
t.Errorf("%s: the waiter's statement reads %q, want it to begin with its family's name line", name, query)
}
var granted bool
if err := d.QueryRowContext(ctx, `SELECT EXISTS (
SELECT 1 FROM pg_locks l, (SELECT hashtextextended($2, 0) AS k) h
WHERE l.locktype = 'advisory' AND l.pid = $1 AND l.granted AND l.objsubid = 1
AND l.classid = ((h.k >> 32) & 4294967295)::oid AND l.objid = (h.k & 4294967295)::oid)`, holderPID, key).Scan(&granted); err != nil {
t.Fatalf("%s: read pg_locks: %v", name, err)
}
if !granted {
t.Errorf("%s: no granted lock on hashtextextended(%q, 0)", name, key)
}
if row.other != nil {
side := dbtest.Tx(t, d).Tx()
if free, err := dbtest.TryLock(ctx, side, row.family, row.other...); err != nil || !free {
t.Errorf("%s: the lock of another id was held (%v)", name, err)
}
}
if err := holder.Rollback(); err != nil {
t.Fatalf("%s: roll back the holder: %v", name, err)
}
if err := <-waiterDone; err != nil {
t.Errorf("%s: the waiter returned %v", name, err)
}
}
// TestSharedHoldersDoNotWait holds that two transactions take the
// materialization family's shared lock without waiting for each other.
func TestSharedHoldersDoNotWait(t *testing.T) {
d := dbtest.Open(t)
ctx := context.Background()
first := dbtest.Tx(t, d).Tx()
if err := db.LockShared(ctx, first, db.Materialization); err != nil {
t.Fatalf("the first shared holder: %v", err)
}
second := dbtest.Tx(t, d)
second.Exec(`SET LOCAL lock_timeout = '1s'`)
if err := db.LockShared(ctx, second.Tx(), db.Materialization); err != nil {
t.Errorf("two shared holders of materialization waited for each other: %v", err)
}
}
// TestUnchangedFamiliesKeepTheirKeys holds that the two families whose keys
// the change kept build the keys the console took before it.
func TestUnchangedFamiliesKeepTheirKeys(t *testing.T) {
d := dbtest.Open(t)
address, err := db.DesiredState.Key(lockUUIDCapitals, "organization", lockUUIDCapitals)
if err != nil {
t.Fatal(err)
}
instance, err := db.Lifecycle.Key(lockUUIDCapitals)
if err != nil {
t.Fatal(err)
}
rows := []struct {
name string
query string
args []any
}{
{"lifecycle", `SELECT hashtextextended('lifecycle:' || $1::uuid::text, 0) = hashtextextended($2, 0)`, []any{lockUUIDCapitals, instance}},
{"desired-state", `SELECT hashtextextended('desired-state:' || $1::uuid::text || ':' || $2 || ':' || $3::uuid::text, 0) = hashtextextended($4, 0)`,
[]any{lockUUIDCapitals, "organization", lockUUIDCapitals, address}},
}
for _, row := range rows {
var same bool
if err := d.QueryRowContext(context.Background(), row.query, row.args...).Scan(&same); err != nil {
t.Fatalf("%s: %v", row.name, err)
}
if !same {
t.Errorf("%s: the family's key differs from the key the console took before", row.name)
}
}
}
+95
View File
@@ -0,0 +1,95 @@
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package db_test
import (
"strings"
"testing"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
)
const (
// lockUUID and lockUUIDCapitals are one UUID in two spellings.
lockUUID = "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"
lockUUIDCapitals = "A0EEBC99-9C0B-4EF8-BB6D-6BB9BD380A11"
)
// keyRow is one family, the parts of an id and either the key the family
// builds or the text its error holds.
type keyRow struct {
name string
family db.Family
id []string
key string
wantErr string
}
// checkKey builds the row's key and compares it, or its error, with what the
// row expects.
func checkKey(t *testing.T, row keyRow) {
t.Helper()
key, err := row.family.Key(row.id...)
switch {
case row.wantErr == "" && err != nil:
t.Errorf("%s: the error is %v, want the key %q", row.name, err, row.key)
case row.wantErr == "" && key != row.key:
t.Errorf("%s: the key is %q, want %q", row.name, key, row.key)
case row.wantErr != "" && err == nil:
t.Errorf("%s: the key is %q with no error, want an error containing %q", row.name, key, row.wantErr)
case row.wantErr != "" && !strings.Contains(err.Error(), row.wantErr):
t.Errorf("%s: the error is %v, want one containing %q", row.name, err, row.wantErr)
}
}
// TestFamilyKeys holds every family's key for its id, and every id a family
// refuses.
func TestFamilyKeys(t *testing.T) {
rows := []keyRow{
{name: "the rendezvous", family: db.Materialization, key: "materialization:"},
{name: "the registry", family: db.DomainsRegistry, key: "domains-registry:"},
{name: "a workspace", family: db.FedWikiWorkspace, id: []string{lockUUID}, key: "fedwiki-workspace:" + lockUUID},
{name: "an instance id in capitals", family: db.Lifecycle, id: []string{lockUUIDCapitals}, key: "lifecycle:" + lockUUID},
{name: "a subscription", family: db.Subscription, id: []string{"sub_123"}, key: "stripe-subscription:sub_123"},
{name: "an invoice", family: db.Invoice, id: []string{"in_123"}, key: "stripe-invoice:in_123"},
{name: "a product", family: db.ProductSync, id: []string{lockUUIDCapitals}, key: "stripe-product-sync:" + lockUUID},
{name: "an address", family: db.DesiredState, id: []string{lockUUIDCapitals, "person", lockUUID}, key: "desired-state:" + lockUUID + ":person:" + lockUUID},
{name: "an empty subscription id", family: db.Subscription, id: []string{""}, wantErr: "id part 1 is empty"},
{name: "a malformed product id", family: db.ProductSync, id: []string{"p1"}, wantErr: "stripe-product-sync: id part 1"},
{name: "a colon in a subscription id", family: db.Subscription, id: []string{"sub:1"}, wantErr: "id part 1 holds ':'"},
{name: "two parts of an address", family: db.DesiredState, id: []string{lockUUID, "person"}, wantErr: "2 id parts, want 3"},
{name: "an id given to the rendezvous", family: db.Materialization, id: []string{lockUUID}, wantErr: "1 id parts, want 0"},
{name: "no family", family: db.Family{}, id: []string{lockUUID}, wantErr: "no family"},
}
for _, row := range rows {
t.Run(strings.ReplaceAll(row.name, " ", "_"), func(t *testing.T) { checkKey(t, row) })
}
}
// TestStatementsNameTheirFamily holds that each family's two statements begin
// with a name line no other family shares, and that Families lists eight
// distinct families.
func TestStatementsNameTheirFamily(t *testing.T) {
const (
exclusive = "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))"
shared = "SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))"
)
firstLines := map[string]string{}
names := map[string]bool{}
for _, f := range db.Families {
names[f.Name()] = true
stmt, sharedStmt := f.Statement(), f.SharedStatement()
if !strings.HasPrefix(stmt, "-- name: ") || !strings.HasSuffix(stmt, exclusive) || !strings.HasSuffix(sharedStmt, shared) {
t.Errorf("%s's statements read %q and %q, want a name line and the exclusive and shared locks", f.Name(), stmt, sharedStmt)
}
first, _, _ := strings.Cut(stmt, "\n")
if other, ok := firstLines[first]; ok {
t.Errorf("%s and %s share the name line %q", other, f.Name(), first)
}
firstLines[first] = f.Name()
}
if len(db.Families) != 8 || len(names) != 8 {
t.Errorf("Families holds %v, want eight distinct names", db.Families)
}
}
@@ -0,0 +1,80 @@
-- SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
-- SPDX-FileCopyrightText: 2025-2026 Christian Galo
-- +goose Up
-- core.assert_rendezvous checks the materialization family's key,
-- hashtextextended('materialization:', 0), which internal/db/locks.go builds
-- for every rendezvous. A backend holding the key this function checked
-- before, a process of the previous release, is refused. A later replacement
-- keeps `v_key` equal to the key `db.Materialization` builds.
-- +goose StatementBegin
CREATE OR REPLACE FUNCTION core.assert_rendezvous(p_exclusive BOOLEAN) RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
v_key BIGINT := hashtextextended('materialization:', 0);
v_held BOOLEAN;
BEGIN
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_locks l
WHERE l.locktype = 'advisory'
AND l.pid = pg_backend_pid()
AND l.granted
AND l.objsubid = 1
AND l.classid = ((v_key >> 32) & 4294967295)::oid
AND l.objid = (v_key & 4294967295)::oid
AND (NOT p_exclusive OR l.mode = 'ExclusiveLock')
) INTO v_held;
IF NOT v_held THEN
RAISE EXCEPTION 'materialization_rendezvous_missing'
USING ERRCODE = '55000',
DETAIL = CASE WHEN p_exclusive
THEN 'the calling transaction does not hold the exclusive materialization rendezvous'
ELSE 'the calling transaction does not hold the materialization rendezvous' END;
END IF;
END;
$$;
ALTER FUNCTION core.assert_rendezvous(BOOLEAN) OWNER TO core_owner;
REVOKE EXECUTE ON FUNCTION core.assert_rendezvous(BOOLEAN) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION core.assert_rendezvous(BOOLEAN) TO core_writer;
-- +goose StatementEnd
-- +goose Down
-- The key the function checked before the materialization family.
-- +goose StatementBegin
CREATE OR REPLACE FUNCTION core.assert_rendezvous(p_exclusive BOOLEAN) RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
v_key BIGINT := hashtextextended('entitlement_materialization', 0);
v_held BOOLEAN;
BEGIN
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_locks l
WHERE l.locktype = 'advisory'
AND l.pid = pg_backend_pid()
AND l.granted
AND l.objsubid = 1
AND l.classid = ((v_key >> 32) & 4294967295)::oid
AND l.objid = (v_key & 4294967295)::oid
AND (NOT p_exclusive OR l.mode = 'ExclusiveLock')
) INTO v_held;
IF NOT v_held THEN
RAISE EXCEPTION 'materialization_rendezvous_missing'
USING ERRCODE = '55000',
DETAIL = CASE WHEN p_exclusive
THEN 'the calling transaction does not hold the exclusive materialization rendezvous'
ELSE 'the calling transaction does not hold the materialization rendezvous' END;
END IF;
END;
$$;
ALTER FUNCTION core.assert_rendezvous(BOOLEAN) OWNER TO core_owner;
REVOKE EXECUTE ON FUNCTION core.assert_rendezvous(BOOLEAN) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION core.assert_rendezvous(BOOLEAN) TO core_writer;
-- +goose StatementEnd
+103
View File
@@ -0,0 +1,103 @@
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package db_test
import (
"context"
"database/sql"
"strings"
"testing"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
)
const (
rendezvousFamilyKeyVersion = 40
rendezvousFamilyKeyPrevious = 39
rendezvousFunction = "core.assert_rendezvous(boolean)"
)
// TestRendezvousFamilyKeyMigration checks, before the migration, after it and
// after its down, which key core.assert_rendezvous accepts, that the
// function's owner and grants never change, and that down restores the
// definition from before the migration.
func TestRendezvousFamilyKeyMigration(t *testing.T) {
scratch := dbtest.Connect(t, dbtest.Scratch(t))
core := db.BaseSources()[0]
queries := dbtest.On(t, scratch)
dbtest.UpTo(t, scratch, core, rendezvousFamilyKeyPrevious)
definition := queries.String(`SELECT pg_get_functiondef($1::regprocedure)`, rendezvousFunction)
privileges := rendezvousFunctionPrivileges(queries)
steps := []struct {
name string
apply bool
up bool
oldKey string
newKey string
sameBody bool
}{
{name: "before the migration", oldKey: "passed", newKey: "refused"},
{name: "after the migration", apply: true, up: true, oldKey: "refused", newKey: "passed"},
{name: "after its down", apply: true, oldKey: "passed", newKey: "refused", sameBody: true},
}
for _, step := range steps {
if step.apply {
dbtest.ApplyVersion(t, scratch, core, rendezvousFamilyKeyVersion, step.up)
}
if got := rendezvousAssertion(t, scratch, func(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, sharedRendezvousBeforeFamilies)
return err
}); got != step.oldKey {
t.Errorf("%s, the assertion %s the old key, want it to be %s", step.name, got, step.oldKey)
}
if got := rendezvousAssertion(t, scratch, func(ctx context.Context, tx *sql.Tx) error {
return db.LockShared(ctx, tx, db.Materialization)
}); got != step.newKey {
t.Errorf("%s, the assertion %s the new key, want it to be %s", step.name, got, step.newKey)
}
if got := rendezvousFunctionPrivileges(queries); got != privileges {
t.Errorf("%s, core.assert_rendezvous's owner or grants read %s, want %s", step.name, got, privileges)
}
if step.sameBody {
if got := queries.String(`SELECT pg_get_functiondef($1::regprocedure)`, rendezvousFunction); got != definition {
t.Errorf("after its down, core.assert_rendezvous reads %q, want the definition before the migration", got)
}
}
}
}
// rendezvousFunctionPrivileges is the owner and the ACL of
// core.assert_rendezvous, which CREATE OR REPLACE must keep.
func rendezvousFunctionPrivileges(queries *dbtest.Q) string {
return queries.String(`SELECT pg_get_userbyid(proowner) || ' ' || COALESCE(proacl::text, '')
FROM pg_proc WHERE oid = $1::regprocedure`, rendezvousFunction)
}
// rendezvousAssertion takes a lock with take in a transaction of its own,
// runs core.assert_rendezvous(false) and reports whether it passed or
// refused with materialization_rendezvous_missing. Any other outcome fails
// the test.
func rendezvousAssertion(t *testing.T, database *sql.DB, take func(context.Context, *sql.Tx) error) string {
t.Helper()
ctx := context.Background()
tx, err := database.BeginTx(ctx, nil)
if err != nil {
t.Fatal(err)
}
defer tx.Rollback()
if err := take(ctx, tx); err != nil {
t.Fatalf("take the lock: %v", err)
}
_, err = tx.ExecContext(ctx, `SELECT core.assert_rendezvous(false)`)
switch {
case err == nil:
return "passed"
case strings.Contains(err.Error(), "materialization_rendezvous_missing"):
return "refused"
}
t.Fatalf("core.assert_rendezvous: %v", err)
return ""
}
+5 -1
View File
@@ -15,6 +15,10 @@ const (
settlementPoolFirstVersion = 38
settlementPoolFirstPrevious = 37
settlementFunction = "core.settle_obligation(uuid,uuid,jsonb,text,integer)"
// sharedRendezvousBeforeFamilies is the key core.assert_rendezvous checks
// at core versions before the materialization family.
sharedRendezvousBeforeFamilies = `SELECT pg_advisory_xact_lock_shared(hashtextextended('entitlement_materialization', 0))`
)
// TestSettlementPoolFirstMigrationUpDownUp checks the restored body, the
@@ -65,7 +69,7 @@ func checkRuntimeSettlement(t *testing.T, database *sql.DB, step string, failure
t.Helper()
queries := dbtest.Tx(t, database)
defer queries.Tx().Rollback()
queries.Exec(sharedRendezvous)
queries.Exec(sharedRendezvousBeforeFamilies)
person := fxPerson(t, queries.Tx())
org := fxOrg(t, queries.Tx(), person)
pool := fxPool(t, queries.Tx(), org, "default")
+12 -22
View File
@@ -54,29 +54,19 @@ const (
ProbeStateError = "error"
)
// registryAllocationLock is the key for the single registry-wide
// pg_advisory_xact_lock every mutating allocation takes (design D3).
// Allocation is human-paced, so one lock for the whole namespace buys the
// subtree-disjointness invariants — which span rows the database cannot
// constrain against each other — at no practical cost.
//
// The value is ASCII "domains", chosen to keep this key space visibly
// distinct from the per-object `hashtext($1)::bigint` locks (see
// docs/database-locks.md): hashtext produces int32-range values, so a 56-bit
// constant can never collide with one.
// registryKind opens a registry allocation's transaction: the registry's
// advisory lock, the domains-registry family's (internal/db/locks.go), is
// its first statement. Allocation is human-paced, so one lock for the whole
// namespace buys the subtree-disjointness invariants, which span rows the
// database cannot constrain against each other, at no practical cost
// (design D3).
//
// LOCK ORDERING INVARIANT: a transaction that touches both the registry and
// entitlement-usage rows MUST take this lock FIRST — the usage row's lock
// (LockNumericUsage, taken by a recount) is always second. A transaction that
// took a usage row and then allocated, and one that allocated and then took
// the usage row, would otherwise grab the two in opposite orders and deadlock.
const registryAllocationLock int64 = 0x646f6d61696e73
// registryKind opens a registry allocation's transaction: the registry's
// advisory lock is its first statement, so nothing in the transaction runs
// before the lock (docs/database-locks.md, Registry).
// entitlement-usage rows MUST take this lock FIRST, as docs/database-locks.md
// orders every advisory family before rows: the usage row's lock
// (LockNumericUsage, taken by a recount) is always second.
var registryKind = db.FirstRuns(func(ctx context.Context, tx *sql.Tx) error {
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock($1)`, registryAllocationLock); err != nil {
if err := db.Lock(ctx, tx, db.DomainsRegistry); err != nil {
return fmt.Errorf("domains: registry advisory lock: %w", err)
}
return nil
@@ -476,8 +466,8 @@ func (t *Tx) PlacementByResource(ctx context.Context, provider, resourceRef stri
// (released on commit or rollback), opened through the runner with
// registryKind. Returning an error rolls the whole composite back; returning
// nil commits. fn's error, a failed begin and a failed commit come back as
// the runner returns them. See registryAllocationLock for the lock-ordering
// invariant against entitlement usage rows.
// the runner returns them. See registryKind for the lock-ordering invariant
// against entitlement usage rows.
func (r *Registry) WithLock(ctx context.Context, fn func(context.Context, *Tx) error) error {
if r.db == nil {
return ErrNoDatabase
+1 -1
View File
@@ -759,7 +759,7 @@ func TestCheckAvailabilityOutsideEveryClaim(t *testing.T) {
// registryLockStatement is the statement the registry's kind runs first,
// named as dbtest.Counted names a raw statement: its first line.
const registryLockStatement = `SELECT pg_advisory_xact_lock($1)`
const registryLockStatement = `LockDomainsRegistry`
// TestRegistryLockIsTheFirstStatement proves a registry transaction runs the
// registry lock before any statement of its composite: the statements of one
@@ -836,7 +836,7 @@ func TestDefaultChangeWaitsForAMoveInFlight(t *testing.T) {
})
changed <- err
}()
dbtest.AwaitLockWait(t, f.w.DB, dbtest.Backend{Query: "pg_advisory_xact_lock(hashtextextended", Waiters: 1})
dbtest.AwaitLockWait(t, f.w.DB, dbtest.Backend{Query: db.Materialization.Statement(), Waiters: 1})
if got := f.q.String(`SELECT default_plan_ladder_id::text FROM core.org_types WHERE org_type = $1`, f.orgType); got != f.next.ID {
t.Errorf("the default is %s while a move is in flight, want the T0 to have saved nothing (Next %s)", got, f.next.ID)
}
@@ -10,6 +10,7 @@ package entitlements_test
// exclusive rendezvous the act's functions assert.
import (
"context"
"database/sql"
"errors"
"maps"
@@ -48,7 +49,9 @@ func newPlanActRows(t *testing.T) planActRows {
// when a test needs the rows visible to another connection.
func newPlanActRowsOn(t *testing.T, q *dbtest.Q) planActRows {
t.Helper()
q.Exec(`SELECT pg_advisory_xact_lock(hashtextextended('entitlement_materialization', 0))`)
if err := db.Lock(context.Background(), q.Tx(), db.Materialization); err != nil {
t.Fatalf("take the rendezvous: %v", err)
}
user := q.ID(`INSERT INTO core.users (oidc_subject) VALUES ($1) RETURNING user_id`, dbtest.Name("pa-subject-"))
person := q.ID(`INSERT INTO core.persons (user_id, display_name, primary_email) VALUES ($1, 'Plan Act', $2) RETURNING person_id`,
user, dbtest.Name("pa-")+"@example.test")
@@ -366,7 +369,9 @@ func TestLockPlanActObligationSurvivesARequeueItWaitedOn(t *testing.T) {
}
defer retrying.Rollback()
rq := dbtest.On(t, retrying)
rq.Exec(`SELECT pg_advisory_xact_lock(hashtextextended('entitlement_materialization', 0))`)
if err := db.Lock(context.Background(), retrying, db.Materialization); err != nil {
t.Fatalf("take the rendezvous: %v", err)
}
retry := planActRows{t: t, q: rq, tx: retrying, e: entitlements.New(retrying), person: r.person}.retry(act)
if n, err := entitlements.New(retrying).RequeuePlanActObligations(ctx, retry); err != nil || n != 1 {
t.Fatalf("requeue: %d, %v", n, err)
@@ -23,6 +23,7 @@ import (
"github.com/google/uuid"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
)
@@ -496,7 +497,7 @@ func TestDeliverDisplaced_WaitsForThePoolRow(t *testing.T) {
t.Fatal(err)
}
defer subLock.Rollback()
if _, err := subLock.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, fx.stripeID); err != nil {
if err := db.Lock(ctx, subLock, db.Subscription, fx.stripeID); err != nil {
t.Fatalf("take S's advisory lock: %v", err)
}
if _, err := subLock.ExecContext(ctx, `UPDATE core.subscriptions SET status = status WHERE subscription_id = $1`, fx.sub); err != nil {
@@ -20,6 +20,7 @@ import (
"time"
"git.coopcloud.tech/wiki-cafe/member-console/internal/billing"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/world"
@@ -341,7 +342,7 @@ func checkEndingActFirst(t *testing.T, name string, end func(f *resumeFixture, t
t.Fatalf("begin the subscription's lock holder: %v", err)
}
defer holder.Rollback()
if _, err := holder.ExecContext(f.w.Ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, "sub_"+f.sub.ID); err != nil {
if err := db.Lock(f.w.Ctx, holder, db.Subscription, "sub_"+f.sub.ID); err != nil {
t.Fatalf("take the subscription's advisory lock: %v", err)
}
if _, err := holder.ExecContext(f.w.Ctx, `UPDATE core.subscriptions SET status = status WHERE subscription_id = $1`, f.sub.ID); err != nil {
+1 -1
View File
@@ -1,7 +1,7 @@
statements:
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
LockMaterializationShared
GetActivePoolProvisionsByPoolID
GetActiveRulesBySetID
SELECT core.assert_rendezvous(false)
+1 -1
View File
@@ -1,7 +1,7 @@
statements:
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock(hashtextextended($1, 0))
LockMaterialization
LockEntitlementSet
GetEntitlementSetRule
GetEntitlementSetRule
+3 -3
View File
@@ -1,7 +1,7 @@
statements:
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock(hashtextextended($1, 0))
LockMaterialization
LockEntitlementSet
GetEntitlementSetRule
GetEntitlementSetRule
@@ -18,7 +18,7 @@ ListPendingObligations
GetObligationStatus
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
LockMaterializationShared
GetObligationStatus
GetObligationStatus
GetObligationStatus
@@ -63,7 +63,7 @@ commit
GetObligationStatus
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
LockMaterializationShared
GetObligationStatus
GetObligationStatus
GetObligationStatus
+12 -13
View File
@@ -17,13 +17,12 @@ import (
)
// The materialization rendezvous (Decision 144): one transaction-scoped
// advisory lock that every transaction materializing a pool holds in shared
// mode and every rule change and operator tier change holds in exclusive
// mode, taken as the transaction's first statement so no holder ever waits
// on something a would-be holder already holds. The demonstration seed adds
// its ladder's tiers on the plain handle, outside it. The key is
// hashtextextended of this name.
const rendezvousName = "entitlement_materialization"
// advisory lock, the materialization family's (internal/db/locks.go), that
// every transaction materializing a pool holds in shared mode and every rule
// change and operator tier change holds in exclusive mode, taken as the
// transaction's first statement so no holder ever waits on something a
// would-be holder already holds. The demonstration seed adds its ladder's
// tiers on the plain handle, outside it.
// rendezvousLockTimeout bounds every wait on the rendezvous; a wait past it
// surfaces as SQLSTATE 55P03 and renders as the refusal copy. It is the lock
@@ -54,7 +53,7 @@ var Materializing = MaterializingWithin(rendezvousLockTimeout, 0)
// transaction timeout of zero sets none. Materializing is the case of 5 s and
// none, so there is one opener for each side of the rendezvous.
func MaterializingWithin(lockTimeout, txTimeout time.Duration) db.Kind {
return db.FirstRuns(rendezvous("pg_advisory_xact_lock_shared", lockTimeout, txTimeout))
return db.FirstRuns(rendezvous(db.LockShared, lockTimeout, txTimeout))
}
// RuleChange opens a transaction that changes a rule or carries an
@@ -65,7 +64,7 @@ func MaterializingWithin(lockTimeout, txTimeout time.Duration) db.Kind {
// and a product's shape in separate statements, so a tier change that
// committed between them would have it classify a tier change against two
// orders of the ladder (#213).
var RuleChange = db.FirstRuns(rendezvous("pg_advisory_xact_lock", rendezvousLockTimeout, 0))
var RuleChange = db.FirstRuns(rendezvous(db.Lock, rendezvousLockTimeout, 0))
// BeginMaterializing opens a Materializing transaction on b.
func BeginMaterializing(ctx context.Context, b db.Beginner) (*sql.Tx, error) {
@@ -78,9 +77,9 @@ func BeginRuleChange(ctx context.Context, b db.Beginner) (*sql.Tx, error) {
}
// rendezvous is the first statements of a transaction that takes the
// materialization rendezvous with lockFn: the lock timeout, the transaction
// timeout when there is one, then the lock.
func rendezvous(lockFn string, lockTimeout, txTimeout time.Duration) func(context.Context, *sql.Tx) error {
// materialization rendezvous with lock, db.Lock or db.LockShared: the lock
// timeout, the transaction timeout when there is one, then the lock.
func rendezvous(lock func(context.Context, *sql.Tx, db.Family, ...string) error, lockTimeout, txTimeout time.Duration) func(context.Context, *sql.Tx) error {
return func(ctx context.Context, tx *sql.Tx) error {
if txTimeout > 0 {
if _, err := tx.ExecContext(ctx, "SET LOCAL transaction_timeout = '"+milliseconds(txTimeout)+"'"); err != nil {
@@ -90,7 +89,7 @@ func rendezvous(lockFn string, lockTimeout, txTimeout time.Duration) func(contex
if _, err := tx.ExecContext(ctx, "SET LOCAL lock_timeout = '"+milliseconds(lockTimeout)+"'"); err != nil {
return fmt.Errorf("set lock_timeout: %w", err)
}
if _, err := tx.ExecContext(ctx, "SELECT "+lockFn+"(hashtextextended($1, 0))", rendezvousName); err != nil {
if err := lock(ctx, tx, db.Materialization); err != nil {
return fmt.Errorf("take materialization rendezvous: %w", err)
}
return nil
+4 -6
View File
@@ -119,9 +119,9 @@ type afterCommit struct {
// reconcileInTx runs the reconcile's transaction on the fetched subscription
// and commits it. Without applyCollection, a collection action the rule asks
// for writes the outbox entry in the transaction.
func reconcileInTx(ctx context.Context, db *sql.DB, logger *slog.Logger, sub *stripe.Subscription, stripeSubscriptionID, reason string, applyCollection bool) (afterCommit, error) {
func reconcileInTx(ctx context.Context, database *sql.DB, logger *slog.Logger, sub *stripe.Subscription, stripeSubscriptionID, reason string, applyCollection bool) (afterCommit, error) {
var plan afterCommit
tx, err := entitlements.BeginMaterializing(ctx, db)
tx, err := entitlements.BeginMaterializing(ctx, database)
if err != nil {
return plan, fmt.Errorf("reconcile: begin tx: %w", err)
}
@@ -129,8 +129,7 @@ func reconcileInTx(ctx context.Context, db *sql.DB, logger *slog.Logger, sub *st
// 1.2 Serialize concurrent reconciles (eager return + webhook) of the same
// subscription so they converge instead of racing the GiST exclusion.
if _, err := tx.ExecContext(ctx,
`SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, stripeSubscriptionID); err != nil {
if err := db.Lock(ctx, tx, db.Subscription, stripeSubscriptionID); err != nil {
return plan, fmt.Errorf("reconcile: advisory lock: %w", err)
}
@@ -1329,8 +1328,7 @@ func supersedeReleasedDowngrades(ctx context.Context, database *sql.DB, stripeSu
return fmt.Errorf("reconcile: begin tx after releasing the schedule: %w", err)
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx,
`SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, stripeSubscriptionID); err != nil {
if err := db.Lock(ctx, tx, db.Subscription, stripeSubscriptionID); err != nil {
return fmt.Errorf("reconcile: advisory lock after releasing the schedule: %w", err)
}
billingQ := billing.New(tx)
@@ -19,6 +19,7 @@ import (
stripe "github.com/stripe/stripe-go/v81"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements"
"git.coopcloud.tech/wiki-cafe/member-console/internal/fulfillment"
"git.coopcloud.tech/wiki-cafe/member-console/internal/stripetest"
@@ -246,8 +247,8 @@ func trackSide(t *testing.T, side *sql.Tx) (*sql.Tx, int) {
// holdSubscriptionLock takes the reconcile's subscription lock on side
// without waiting, and fails when another transaction holds it.
func holdSubscriptionLock(ctx context.Context, side *sql.Tx, stripeSubID string) error {
var free bool
if err := side.QueryRowContext(ctx, `SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint)`, stripeSubID).Scan(&free); err != nil {
free, err := dbtest.TryLock(ctx, side, db.Subscription, stripeSubID)
if err != nil {
return fmt.Errorf("try the subscription lock: %w", err)
}
if !free {
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
rollback
== logs
== rows
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -35,7 +35,7 @@ commit
stripe POST /v1/subscription_schedules/sub_sched_mock/release
probe: the subscription lock is free
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockSubscription
SupersedeScheduledChange
commit
== logs
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -5,8 +5,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -9,8 +9,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -9,8 +9,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -10,8 +10,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -10,8 +10,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -10,8 +10,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -10,8 +10,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -10,8 +10,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
@@ -15,8 +15,8 @@ GetSubscriptionMappingByStripeID
stripe GET /v1/subscriptions/<stripe-subscription> expand[0]=items.data.price&expand[1]=schedule
begin
SET LOCAL lock_timeout = '5s'
SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockMaterializationShared
LockSubscription
GetCustomerMappingByStripeCustomerID
GetBillingAccountByID
GetDefaultPoolByOrgID
+2 -2
View File
@@ -228,8 +228,8 @@ func (e *traceEnv) probeSubscriptionLock() string {
e.t.Fatalf("probe: begin: %v", err)
}
defer func() { _ = tx.Rollback() }()
var free bool
if err := tx.QueryRowContext(e.ctx, `SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint)`, e.subID).Scan(&free); err != nil {
free, err := dbtest.TryLock(e.ctx, tx, db.Subscription, e.subID)
if err != nil {
e.t.Fatalf("probe: try the lock: %v", err)
}
if free {
@@ -20,6 +20,7 @@ import (
"github.com/google/uuid"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/integration"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
)
@@ -468,8 +469,7 @@ func TestDesiredStateRecordQueries_DB(t *testing.T) {
pool := d.pool(org)
d.Exec(`INSERT INTO core.numeric_entitlements (pool_id, resource_key, entitlement_type, resource_limit) VALUES ($1, $2, 'limit', 3)`, pool, r.sites)
if err := d.q.LockDesiredStateAddress(d.ctx, integration.LockDesiredStateAddressParams{
ConnectionID: r.conn, RecipientKind: "organization", RecipientID: org}); err != nil {
if err := db.Lock(d.ctx, d.tx, db.DesiredState, r.conn, "organization", org); err != nil {
t.Fatal(err)
}
addr := integration.LockDesiredStateRecordParams{ConnectionID: r.conn, OrgID: nullUUID(org)}
@@ -665,7 +665,7 @@ func newLRQ(t *testing.T) *lrq {
// insert takes the instance's lock and inserts a request at generation g.
func (l *lrq) insert(g int64, op, key, params string) (string, error) {
l.t.Helper()
if err := l.q.LockLifecycleInstance(l.ctx, l.instance); err != nil {
if err := db.Lock(l.ctx, l.tx, db.Lifecycle, l.instance); err != nil {
l.t.Fatal(err)
}
return l.q.InsertLifecycleRequest(l.ctx, integration.InsertLifecycleRequestParams{
+9 -28
View File
@@ -836,35 +836,8 @@ func (q *Queries) ListLaggingDesiredStateRecords(ctx context.Context, arg ListLa
return items, nil
}
const lockDesiredStateAddress = `-- name: LockDesiredStateAddress :exec
SELECT pg_advisory_xact_lock(hashtextextended(
'desired-state:' || $1::uuid::text
|| ':' || $2::text
|| ':' || $3::uuid::text, 0))
`
type LockDesiredStateAddressParams struct {
ConnectionID string `json:"connection_id"`
RecipientKind string `json:"recipient_kind"`
RecipientID string `json:"recipient_id"`
}
// ---------------------------------------------------------------------------
// Recomputing one record (design D9), in one READ COMMITTED transaction, in
// this order: LockDesiredStateAddress, LockDesiredStateRecord, then
// DesiredStateContent in a new statement, then UpdateDesiredStateRecord or
// InsertDesiredStateRecord. Every transaction that writes a record's content
// or generation, or a projection, takes the address lock first.
// ---------------------------------------------------------------------------
// The serialization point for one (connection, recipient) address, including
// one that has no record yet. recipient_kind is 'person' or 'organization'.
func (q *Queries) LockDesiredStateAddress(ctx context.Context, arg LockDesiredStateAddressParams) error {
_, err := q.db.ExecContext(ctx, lockDesiredStateAddress, arg.ConnectionID, arg.RecipientKind, arg.RecipientID)
return err
}
const lockDesiredStateRecord = `-- name: LockDesiredStateRecord :one
SELECT record_id, connection_id, person_id, org_id, content, content_hash, generation, generation_at, last_dispatched_at, last_push_error, last_push_failed_at, created_at, updated_at FROM core.desired_state_records r
WHERE (r.connection_id = $1::uuid AND r.person_id = $2::uuid)
OR (r.connection_id = $1::uuid AND r.org_id = $3::uuid)
@@ -877,6 +850,14 @@ type LockDesiredStateRecordParams struct {
OrgID uuid.NullUUID `json:"org_id"`
}
// ---------------------------------------------------------------------------
// Recomputing one record (design D9), in one READ COMMITTED transaction, in
// this order: the desired-state family's lock (db.Lock),
// LockDesiredStateRecord, then DesiredStateContent in a new statement, then
// UpdateDesiredStateRecord or InsertDesiredStateRecord. Every transaction that
// writes a record's content or generation, or a projection, takes the address
// lock first.
// ---------------------------------------------------------------------------
// The record at one address, locked; exactly one of person_id and org_id is
// set. sql.ErrNoRows when the address has no record.
func (q *Queries) LockDesiredStateRecord(ctx context.Context, arg LockDesiredStateRecordParams) (DesiredStateRecord, error) {
+5
View File
@@ -25,3 +25,8 @@ func (k LifecycleAnswerKind) String() string {
return fmt.Sprintf("LifecycleAnswerKind(%d)", int(k))
}
}
// SetAfterInstanceLock sets a hook that runs inside a recording right after
// its instance lock is granted, and clears it when fn is nil; a test that
// sets it resets it in cleanup.
func SetAfterInstanceLock(fn func(string)) { afterInstanceLock = fn }
+74 -55
View File
@@ -114,10 +114,19 @@ func NewLifecycleGeneration(ctx context.Context, db DBTX) (int64, error) {
return New(db).MintLifecycleGeneration(ctx)
}
// RecordLifecycleRequest records spec or answers from what holds its address
// (design D26). It serializes per instance through a transaction-scoped
// advisory lock, so dbtx must be a transaction for the answer to hold until
// the caller commits; given a *sql.DB it runs in a transaction of its own.
// RecordLifecycleRequest records spec, or answers from what holds its
// address, in a transaction of its own on b (a *sql.DB, or a *sql.Conn a
// test pins), committed when it answers. RecordLifecycleRequestInTx records
// in the caller's transaction.
func RecordLifecycleRequest(ctx context.Context, b db.Beginner, spec LifecycleRequestSpec) (LifecycleAnswer, error) {
return inOwnTx(ctx, b, func(tx *sql.Tx) (LifecycleAnswer, error) {
return RecordLifecycleRequestInTx(ctx, tx, spec)
})
}
// RecordLifecycleRequestInTx records spec or answers from what holds its
// address (design D26), in tx, which the caller commits; the lifecycle
// family's lock serializes it per instance until then.
//
// Under the lock it reads the instance's latest generation. A generation
// above it is new: the request is recorded unless another request on the
@@ -125,51 +134,57 @@ func NewLifecycleGeneration(ctx context.Context, db DBTX) (int64, error) {
// core.generation_ceiling() (LifecycleConflict). A generation at or below it is a repeat
// when the address holds the same content, and a conflict otherwise, which
// also covers an older tab's generation that was never recorded.
func RecordLifecycleRequest(ctx context.Context, dbtx DBTX, spec LifecycleRequestSpec) (LifecycleAnswer, error) {
return inOwnTx(ctx, dbtx, func(tx DBTX) (LifecycleAnswer, error) {
if err := spec.checkAddress("record lifecycle request"); err != nil {
return LifecycleAnswer{}, err
}
content, err := spec.content()
if err != nil {
return LifecycleAnswer{}, err
}
q := New(tx)
latest, err := lockLatest(ctx, q, spec.InstanceID)
if err != nil {
return LifecycleAnswer{}, err
}
return decide(ctx, q, spec, content, latest)
})
func RecordLifecycleRequestInTx(ctx context.Context, tx *sql.Tx, spec LifecycleRequestSpec) (LifecycleAnswer, error) {
if err := spec.checkAddress("record lifecycle request"); err != nil {
return LifecycleAnswer{}, err
}
content, err := spec.content()
if err != nil {
return LifecycleAnswer{}, err
}
q := New(tx)
latest, err := lockLatest(ctx, tx, q, spec.InstanceID)
if err != nil {
return LifecycleAnswer{}, err
}
return decide(ctx, q, spec, content, latest)
}
// RecordFirstLifecycleRequest records a request that must be its instance's
// first, such as a create, or answers from what holds its address. It runs
// as RecordLifecycleRequest does, except that a generation above the
// instance's latest is a conflict when the instance already holds a request:
// a create posted again under another generation must not become a second
// create of the same instance.
func RecordFirstLifecycleRequest(ctx context.Context, dbtx DBTX, spec LifecycleRequestSpec) (LifecycleAnswer, error) {
return inOwnTx(ctx, dbtx, func(tx DBTX) (LifecycleAnswer, error) {
if err := spec.checkAddress("record first lifecycle request"); err != nil {
return LifecycleAnswer{}, err
}
q := New(tx)
latest, err := lockLatest(ctx, q, spec.InstanceID)
if err != nil {
return LifecycleAnswer{}, err
}
if latest != 0 && spec.Generation > latest {
return LifecycleAnswer{Kind: LifecycleConflict}, nil
}
content, err := spec.content()
if err != nil {
return LifecycleAnswer{}, err
}
return decide(ctx, q, spec, content, latest)
// first, such as a create, or answers from what holds its address, in a
// transaction of its own on b, as RecordLifecycleRequest does.
// RecordFirstLifecycleRequestInTx records in the caller's transaction.
func RecordFirstLifecycleRequest(ctx context.Context, b db.Beginner, spec LifecycleRequestSpec) (LifecycleAnswer, error) {
return inOwnTx(ctx, b, func(tx *sql.Tx) (LifecycleAnswer, error) {
return RecordFirstLifecycleRequestInTx(ctx, tx, spec)
})
}
// RecordFirstLifecycleRequestInTx records a request that must be its
// instance's first, such as a create, or answers from what holds its
// address. It runs as RecordLifecycleRequestInTx does, except that a
// generation above the instance's latest is a conflict when the instance
// already holds a request: a create posted again under another generation
// must not become a second create of the same instance.
func RecordFirstLifecycleRequestInTx(ctx context.Context, tx *sql.Tx, spec LifecycleRequestSpec) (LifecycleAnswer, error) {
if err := spec.checkAddress("record first lifecycle request"); err != nil {
return LifecycleAnswer{}, err
}
q := New(tx)
latest, err := lockLatest(ctx, tx, q, spec.InstanceID)
if err != nil {
return LifecycleAnswer{}, err
}
if latest != 0 && spec.Generation > latest {
return LifecycleAnswer{Kind: LifecycleConflict}, nil
}
content, err := spec.content()
if err != nil {
return LifecycleAnswer{}, err
}
return decide(ctx, q, spec, content, latest)
}
// decide answers a spec under its instance's lock, given the latest
// generation read under it.
func decide(ctx context.Context, q *Queries, spec LifecycleRequestSpec, c requestContent, latest int64) (LifecycleAnswer, error) {
@@ -179,16 +194,12 @@ func decide(ctx context.Context, q *Queries, spec LifecycleRequestSpec, c reques
return answerAddress(ctx, q, spec, c)
}
// inOwnTx runs fn on dbtx, or, when dbtx is a *sql.DB, in a transaction of
// its own that commits when fn succeeds. The answer comes back with a
// failed commit's error, as each caller returned it before.
func inOwnTx(ctx context.Context, dbtx DBTX, fn func(DBTX) (LifecycleAnswer, error)) (LifecycleAnswer, error) {
sqlDB, ok := dbtx.(*sql.DB)
if !ok {
return fn(dbtx)
}
// inOwnTx runs fn in a transaction of its own on b, committed when fn
// succeeds. The answer comes back with a failed commit's error, as each
// caller returned it before.
func inOwnTx(ctx context.Context, b db.Beginner, fn func(*sql.Tx) (LifecycleAnswer, error)) (LifecycleAnswer, error) {
var answer LifecycleAnswer
err := db.InTx(ctx, sqlDB, db.Plain, func(tx *sql.Tx) error {
err := db.InTx(ctx, b, db.Plain, func(tx *sql.Tx) error {
var err error
answer, err = fn(tx)
return err
@@ -229,11 +240,19 @@ func (spec LifecycleRequestSpec) content() (requestContent, error) {
return c, nil
}
// lockLatest takes the instance's lock and reads its latest generation.
func lockLatest(ctx context.Context, q *Queries, instanceID string) (int64, error) {
if err := q.LockLifecycleInstance(ctx, instanceID); err != nil {
// afterInstanceLock, when set by a test, runs inside a recording right after
// the instance's lock is granted and before its latest generation is read.
var afterInstanceLock func(instanceID string)
// lockLatest takes the instance's lock on tx and reads its latest
// generation.
func lockLatest(ctx context.Context, tx *sql.Tx, q *Queries, instanceID string) (int64, error) {
if err := db.Lock(ctx, tx, db.Lifecycle, instanceID); err != nil {
return 0, fmt.Errorf("lock lifecycle instance: %w", err)
}
if afterInstanceLock != nil {
afterInstanceLock(instanceID)
}
latest, err := q.LatestLifecycleGeneration(ctx, instanceID)
if err != nil {
return 0, fmt.Errorf("latest lifecycle generation: %w", err)
@@ -145,9 +145,9 @@ func recordCase(d *dq, f lifecycleFixture, statements *dbtest.Statements, norm *
if c.edit != nil {
c.edit(f, d, &spec)
}
record := integration.RecordLifecycleRequest
record := integration.RecordLifecycleRequestInTx
if c.first {
record = integration.RecordFirstLifecycleRequest
record = integration.RecordFirstLifecycleRequestInTx
}
statements.Reset()
a, err := record(d.ctx, d.tx, spec)
+53 -7
View File
@@ -21,6 +21,7 @@ import (
"github.com/google/uuid"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/integration"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
@@ -59,7 +60,7 @@ func (d *dq) generation() int64 {
func (d *dq) record(spec integration.LifecycleRequestSpec) integration.LifecycleAnswer {
d.t.Helper()
a, err := integration.RecordLifecycleRequest(d.ctx, d.tx, spec)
a, err := integration.RecordLifecycleRequestInTx(d.ctx, d.tx, spec)
if err != nil {
d.t.Fatalf("record %+v: %v", spec, err)
}
@@ -197,7 +198,7 @@ func TestRecordLifecycleRequestOlderTabAndFloors_DB(t *testing.T) {
foreignKey.ResourceKey = "external_domain_claims"
for name, s := range map[string]integration.LifecycleRequestSpec{"operation": undeclared, "key": foreignKey} {
d.Exec("SAVEPOINT refused")
if _, err := integration.RecordLifecycleRequest(d.ctx, d.tx, s); !errors.Is(err, integration.ErrLifecycleOperationRefused) {
if _, err := integration.RecordLifecycleRequestInTx(d.ctx, d.tx, s); !errors.Is(err, integration.ErrLifecycleOperationRefused) {
t.Errorf("undeclared %s: %v", name, err)
}
d.Exec("ROLLBACK TO SAVEPOINT refused")
@@ -330,7 +331,7 @@ func TestRecordLifecycleRequestConcurrentOlderTab_DB(t *testing.T) {
t.Fatal(err)
}
defer newer.Rollback()
if a, err := integration.RecordLifecycleRequest(ctx, newer, fx.spec(instance, g2, integration.LifecycleDelete, `{}`)); err != nil || a.Kind != integration.LifecycleRecorded {
if a, err := integration.RecordLifecycleRequestInTx(ctx, newer, fx.spec(instance, g2, integration.LifecycleDelete, `{}`)); err != nil || a.Kind != integration.LifecycleRecorded {
t.Fatalf("newer = %+v, %v", a, err)
}
@@ -354,7 +355,7 @@ func TestRecordLifecycleRequestConcurrentOlderTab_DB(t *testing.T) {
done <- result{err: err}
return
}
a, err := integration.RecordLifecycleRequest(ctx, tx, fx.spec(instance, g1, integration.LifecycleDelete, `{}`))
a, err := integration.RecordLifecycleRequestInTx(ctx, tx, fx.spec(instance, g1, integration.LifecycleDelete, `{}`))
if err == nil {
err = tx.Commit()
} else {
@@ -392,7 +393,7 @@ func TestRecordLifecycleRequestOwnTransaction_DB(t *testing.T) {
counted, statements := dbtest.Counted(t)
ctx := context.Background()
fx := committedLifecycleFixture(t, database)
record := map[string]func(context.Context, integration.DBTX, integration.LifecycleRequestSpec) (integration.LifecycleAnswer, error){
record := map[string]func(context.Context, db.Beginner, integration.LifecycleRequestSpec) (integration.LifecycleAnswer, error){
"RecordLifecycleRequest": integration.RecordLifecycleRequest,
"RecordFirstLifecycleRequest": integration.RecordFirstLifecycleRequest,
}
@@ -557,7 +558,7 @@ func TestFirstLifecycleRequestRepeatAndWorkspace_DB(t *testing.T) {
instance := uuid.New().String()
spec := f.spec(instance, d.generation(), integration.LifecycleCreate, `{"domain": "first.example.test"}`)
for i, want := range []integration.LifecycleAnswerKind{integration.LifecycleRecorded, integration.LifecycleRepeatOpen} {
a, err := integration.RecordFirstLifecycleRequest(d.ctx, d.tx, spec)
a, err := integration.RecordFirstLifecycleRequestInTx(d.ctx, d.tx, spec)
if err != nil || a.Kind != want {
t.Fatalf("first request, call %d = %v, %v; want %v", i+1, a.Kind, err, want)
}
@@ -584,7 +585,7 @@ func TestLifecycleGenerationAboveTheCeilingIsRefused_DB(t *testing.T) {
if a := d.record(f.spec(instance, math.MaxInt64, integration.LifecycleCreate, `{"domain": "top.example.test"}`)); a.Kind != integration.LifecycleConflict {
t.Errorf("a generation of MaxInt64 = %+v, want a conflict", a)
}
if a, err := integration.RecordFirstLifecycleRequest(d.ctx, d.tx, f.spec(instance, math.MaxInt64, integration.LifecycleCreate, `{}`)); err != nil || a.Kind != integration.LifecycleConflict {
if a, err := integration.RecordFirstLifecycleRequestInTx(d.ctx, d.tx, f.spec(instance, math.MaxInt64, integration.LifecycleCreate, `{}`)); err != nil || a.Kind != integration.LifecycleConflict {
t.Errorf("a first request at MaxInt64 = %+v, %v; want a conflict", a, err)
}
if n := d.requestRows(instance); n != 0 {
@@ -603,3 +604,48 @@ func TestLifecycleGenerationAboveTheCeilingIsRefused_DB(t *testing.T) {
}
d.Exec(`ROLLBACK TO SAVEPOINT above_ceiling`)
}
// TestARecordingOnAConnectionHoldsTheInstanceLock records on a connection a
// test pins, which the recording function used to run on without a
// transaction, so the instance lock ended with the statement that took it. A
// hook that runs right after the lock is granted probes the lock from another
// transaction.
func TestARecordingOnAConnectionHoldsTheInstanceLock(t *testing.T) {
database := dbtest.Open(t)
ctx := context.Background()
fx := committedLifecycleFixture(t, database)
conn, err := database.Conn(ctx)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = conn.Close() })
instance := uuid.New().String()
gen, err := integration.NewLifecycleGeneration(ctx, database)
if err != nil {
t.Fatal(err)
}
var probed, free bool
var probeErr error
integration.SetAfterInstanceLock(func(id string) {
side, err := database.BeginTx(ctx, nil)
if err != nil {
probeErr = err
return
}
defer side.Rollback()
probed = true
free, probeErr = dbtest.TryLock(ctx, side, db.Lifecycle, id)
})
t.Cleanup(func() { integration.SetAfterInstanceLock(nil) })
a, err := integration.RecordLifecycleRequest(ctx, conn, fx.spec(instance, gen, integration.LifecycleCreate, `{}`))
if err != nil || a.Kind != integration.LifecycleRecorded {
t.Errorf("the recording on a pinned connection answered %v, %v, want accepted", a, err)
}
if probeErr != nil || !probed {
t.Fatalf("the probe after the instance lock ran %v, failed with %v", probed, probeErr)
}
if free {
t.Errorf("a recording on a pinned connection left the instance lock free after taking it")
}
}
+9 -17
View File
@@ -262,6 +262,12 @@ FROM core.lifecycle_requests
WHERE instance_id = $1::uuid
`
// Lifecycle requests (design D24 to D28). Recording a request runs, in one
// transaction: the lifecycle family's lock (db.Lock),
// LatestLifecycleGeneration, then for a generation above the latest
// GenerationCeiling, OpenLifecycleGeneration and InsertLifecycleRequest, and
// otherwise GetLifecycleRequestByAddress, so the latest-generation check and
// the insert are one step per instance.
// The highest generation recorded for an instance, 0 when it has none.
func (q *Queries) LatestLifecycleGeneration(ctx context.Context, instanceID string) (int64, error) {
row := q.db.QueryRowContext(ctx, latestLifecycleGeneration, instanceID)
@@ -326,20 +332,6 @@ func (q *Queries) ListAcceptedLifecycleRequestsOlderThan(ctx context.Context, ar
return items, nil
}
const lockLifecycleInstance = `-- name: LockLifecycleInstance :exec
SELECT pg_advisory_xact_lock(hashtextextended('lifecycle:' || $1::uuid::text, 0))
`
// Lifecycle requests (design D24 to D28). Recording a request runs, in one
// transaction: LockLifecycleInstance, LatestLifecycleGeneration, then for a
// generation above the latest GenerationCeiling, OpenLifecycleGeneration and
// InsertLifecycleRequest, and otherwise GetLifecycleRequestByAddress, so the
// latest-generation check and the insert are one step per instance.
func (q *Queries) LockLifecycleInstance(ctx context.Context, instanceID string) error {
_, err := q.db.ExecContext(ctx, lockLifecycleInstance, instanceID)
return err
}
const mintInstanceID = `-- name: MintInstanceID :one
SELECT uuidv7()::uuid AS instance_id
`
@@ -411,9 +403,9 @@ WHERE instance_id = $1::uuid AND status = 'accepted'
`
// The generation of the instance's accepted request, 0 when none is open.
// Read under LockLifecycleInstance, so the one-open rule is answered before
// the insert instead of by a unique violation that would abort the caller's
// transaction.
// Read under the lifecycle family's lock (db.Lock), so the one-open rule is
// answered before the insert instead of by a unique violation that would
// abort the caller's transaction.
func (q *Queries) OpenLifecycleGeneration(ctx context.Context, instanceID string) (int64, error) {
row := q.db.QueryRowContext(ctx, openLifecycleGeneration, instanceID)
var open_generation int64
+14 -16
View File
@@ -145,6 +145,12 @@ type Querier interface {
// accepted request on the instance under another generation raises a unique
// violation on uq_lifecycle_requests_one_open.
InsertLifecycleRequest(ctx context.Context, arg InsertLifecycleRequestParams) (string, error)
// Lifecycle requests (design D24 to D28). Recording a request runs, in one
// transaction: the lifecycle family's lock (db.Lock),
// LatestLifecycleGeneration, then for a generation above the latest
// GenerationCeiling, OpenLifecycleGeneration and InsertLifecycleRequest, and
// otherwise GetLifecycleRequestByAddress, so the latest-generation check and
// the insert are one step per instance.
// The highest generation recorded for an instance, 0 when it has none.
LatestLifecycleGeneration(ctx context.Context, instanceID string) (int64, error)
// The boot pass (design D27): accepted requests at one connection older than
@@ -191,23 +197,15 @@ type Querier interface {
ListSubscriptionsWithRecentReconcileEntries(ctx context.Context, subscriptionIds []string) ([]string, error)
// ---------------------------------------------------------------------------
// Recomputing one record (design D9), in one READ COMMITTED transaction, in
// this order: LockDesiredStateAddress, LockDesiredStateRecord, then
// DesiredStateContent in a new statement, then UpdateDesiredStateRecord or
// InsertDesiredStateRecord. Every transaction that writes a record's content
// or generation, or a projection, takes the address lock first.
// this order: the desired-state family's lock (db.Lock),
// LockDesiredStateRecord, then DesiredStateContent in a new statement, then
// UpdateDesiredStateRecord or InsertDesiredStateRecord. Every transaction that
// writes a record's content or generation, or a projection, takes the address
// lock first.
// ---------------------------------------------------------------------------
// The serialization point for one (connection, recipient) address, including
// one that has no record yet. recipient_kind is 'person' or 'organization'.
LockDesiredStateAddress(ctx context.Context, arg LockDesiredStateAddressParams) error
// The record at one address, locked; exactly one of person_id and org_id is
// set. sql.ErrNoRows when the address has no record.
LockDesiredStateRecord(ctx context.Context, arg LockDesiredStateRecordParams) (DesiredStateRecord, error)
// Lifecycle requests (design D24 to D28). Recording a request runs, in one
// transaction: LockLifecycleInstance, LatestLifecycleGeneration, then for a
// generation above the latest GenerationCeiling, OpenLifecycleGeneration and
// InsertLifecycleRequest, and otherwise GetLifecycleRequestByAddress, so the
// latest-generation check and the insert are one step per instance.
LockLifecycleInstance(ctx context.Context, instanceID string) error
// A failed attempt on a row: for a wide row, its expansion query failed; for
// a recipient_only row, its recompute failed. The row becomes failed (the
// dead letter) when attempts reaches max_attempts and otherwise stays
@@ -229,9 +227,9 @@ type Querier interface {
// request and at least the clock, one row per instance asked for.
NextLifecycleGenerations(ctx context.Context, instanceIds []string) ([]NextLifecycleGenerationsRow, error)
// The generation of the instance's accepted request, 0 when none is open.
// Read under LockLifecycleInstance, so the one-open rule is answered before
// the insert instead of by a unique violation that would abort the caller's
// transaction.
// Read under the lifecycle family's lock (db.Lock), so the one-open rule is
// answered before the insert instead of by a unique violation that would
// abort the caller's transaction.
OpenLifecycleGeneration(ctx context.Context, instanceID string) (int64, error)
// What a workspace's list shows for one connection: every accepted request,
// and per instance the latest request when it is declined or failed and not
+5 -12
View File
@@ -252,20 +252,13 @@ LIMIT sqlc.arg(page_size)::int;
-- ---------------------------------------------------------------------------
-- Recomputing one record (design D9), in one READ COMMITTED transaction, in
-- this order: LockDesiredStateAddress, LockDesiredStateRecord, then
-- DesiredStateContent in a new statement, then UpdateDesiredStateRecord or
-- InsertDesiredStateRecord. Every transaction that writes a record's content
-- or generation, or a projection, takes the address lock first.
-- this order: the desired-state family's lock (db.Lock),
-- LockDesiredStateRecord, then DesiredStateContent in a new statement, then
-- UpdateDesiredStateRecord or InsertDesiredStateRecord. Every transaction that
-- writes a record's content or generation, or a projection, takes the address
-- lock first.
-- ---------------------------------------------------------------------------
-- name: LockDesiredStateAddress :exec
-- The serialization point for one (connection, recipient) address, including
-- one that has no record yet. recipient_kind is 'person' or 'organization'.
SELECT pg_advisory_xact_lock(hashtextextended(
'desired-state:' || sqlc.arg(connection_id)::uuid::text
|| ':' || sqlc.arg(recipient_kind)::text
|| ':' || sqlc.arg(recipient_id)::uuid::text, 0));
-- name: LockDesiredStateRecord :one
-- The record at one address, locked; exactly one of person_id and org_id is
-- set. sql.ErrNoRows when the address has no record.
@@ -1,12 +1,10 @@
-- name: LockLifecycleInstance :exec
-- Lifecycle requests (design D24 to D28). Recording a request runs, in one
-- transaction: LockLifecycleInstance, LatestLifecycleGeneration, then for a
-- generation above the latest GenerationCeiling, OpenLifecycleGeneration and
-- InsertLifecycleRequest, and otherwise GetLifecycleRequestByAddress, so the
-- latest-generation check and the insert are one step per instance.
SELECT pg_advisory_xact_lock(hashtextextended('lifecycle:' || sqlc.arg(instance_id)::uuid::text, 0));
-- name: LatestLifecycleGeneration :one
-- Lifecycle requests (design D24 to D28). Recording a request runs, in one
-- transaction: the lifecycle family's lock (db.Lock),
-- LatestLifecycleGeneration, then for a generation above the latest
-- GenerationCeiling, OpenLifecycleGeneration and InsertLifecycleRequest, and
-- otherwise GetLifecycleRequestByAddress, so the latest-generation check and
-- the insert are one step per instance.
-- The highest generation recorded for an instance, 0 when it has none.
SELECT COALESCE(MAX(generation), 0)::bigint AS latest_generation
FROM core.lifecycle_requests
@@ -128,9 +126,9 @@ GROUP BY ids.instance_id;
-- name: OpenLifecycleGeneration :one
-- The generation of the instance's accepted request, 0 when none is open.
-- Read under LockLifecycleInstance, so the one-open rule is answered before
-- the insert instead of by a unique violation that would abort the caller's
-- transaction.
-- Read under the lifecycle family's lock (db.Lock), so the one-open rule is
-- answered before the insert instead of by a unique violation that would
-- abort the caller's transaction.
SELECT COALESCE(MAX(generation), 0)::bigint AS open_generation
FROM core.lifecycle_requests
WHERE instance_id = sqlc.arg(instance_id)::uuid AND status = 'accepted';
@@ -33,24 +33,25 @@ func RecordSiteRequest(ctx context.Context, database db.Beginner, spec integrati
// RecordSiteRequestInTx validates new status intent under the instance lock.
// The caller supplies a READ COMMITTED transaction and rolls back on refusal.
func RecordSiteRequestInTx(ctx context.Context, tx *sql.Tx, spec integration.LifecycleRequestSpec, domain string) (integration.LifecycleAnswer, error) {
return recordSiteRequestInTx(ctx, tx, spec, domain)
return recordSiteRequestInTx(ctx, tx, tx, spec, domain)
}
func recordSiteRequestInTx(ctx context.Context, tx DBTX, spec integration.LifecycleRequestSpec, domain string) (integration.LifecycleAnswer, error) {
queries := integration.New(tx)
if err := queries.LockLifecycleInstance(ctx, spec.InstanceID); err != nil {
// recordSiteRequestInTx takes the instance lock and records on tx, and runs
// every read on q, a handle a test wraps to make one of them fail.
func recordSiteRequestInTx(ctx context.Context, tx *sql.Tx, q DBTX, spec integration.LifecycleRequestSpec, domain string) (integration.LifecycleAnswer, error) {
if err := db.Lock(ctx, tx, db.Lifecycle, spec.InstanceID); err != nil {
return integration.LifecycleAnswer{}, fmt.Errorf("lock lifecycle instance: %w", err)
}
_, _, found, err := integration.LookupLifecycleRequest(ctx, tx, spec.WorkspaceID, spec.InstanceID, spec.Generation)
_, _, found, err := integration.LookupLifecycleRequest(ctx, q, spec.WorkspaceID, spec.InstanceID, spec.Generation)
if err != nil {
return integration.LifecycleAnswer{}, err
}
if !found {
if err := checkRequestSite(ctx, tx, spec.InstanceID, spec.WorkspaceID, domain); err != nil {
if err := checkRequestSite(ctx, q, spec.InstanceID, spec.WorkspaceID, domain); err != nil {
return integration.LifecycleAnswer{}, err
}
}
return integration.RecordLifecycleRequest(ctx, tx, spec)
return integration.RecordLifecycleRequestInTx(ctx, tx, spec)
}
// checkRequestSite reads an original site under an instance lock the caller holds.
@@ -72,7 +73,7 @@ func checkRequestSite(ctx context.Context, tx DBTX, instanceID, workspaceID, dom
// The caller supplies a READ COMMITTED transaction and records activation then park.
func CheckSwapRequestSites(ctx context.Context, tx *sql.Tx, workspaceID, activateID, activateDomain, parkID, parkDomain string) error {
for _, instanceID := range []string{activateID, parkID} {
if err := integration.New(tx).LockLifecycleInstance(ctx, instanceID); err != nil {
if err := db.Lock(ctx, tx, db.Lifecycle, instanceID); err != nil {
return err
}
}
@@ -88,7 +88,7 @@ func checkIntakeOrder(t *testing.T, row intakeOrderCase) {
}
defer holder.Rollback()
pid := int(dbtest.On(t, holder).Int("SELECT pg_backend_pid()"))
if err := integration.New(holder).LockLifecycleInstance(env.Ctx, instance); err != nil {
if err := db.Lock(env.Ctx, holder, db.Lifecycle, instance); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(env.Ctx, 10*time.Second)
@@ -154,7 +154,7 @@ func checkProtectedReadError(t *testing.T, database *sql.DB, spec integration.Li
t.Fatal(err)
}
defer tx.Rollback()
_, err = fwmod.RecordSiteRequestOn(t.Context(), failedSiteRead{tx}, spec, domain)
_, err = fwmod.RecordSiteRequestOn(t.Context(), tx, failedSiteRead{tx}, spec, domain)
var missing *fwmod.MissingSiteError
if err == nil || errors.As(err, &missing) {
t.Errorf("%s: protected read hid a database error as not found", name)
@@ -192,7 +192,7 @@ func TestSiteRequestRepeats_DB(t *testing.T) {
// deleteWithoutRequest checks that the locked read sees committed intake.
func deleteWithoutRequest(ctx context.Context, intakeDB *sql.DB, instance string) error {
return db.InTx(ctx, intakeDB, db.ReadCommitted, func(tx *sql.Tx) error {
if err := integration.New(tx).LockLifecycleInstance(ctx, instance); err != nil {
if err := db.Lock(ctx, tx, db.Lifecycle, instance); err != nil {
return err
}
open, err := fwmod.New(tx).SiteHasOpenRequest(ctx, instance)
@@ -216,7 +216,7 @@ func recordIntakePair(ctx context.Context, database *sql.DB, spec integration.Li
if err := fwmod.CheckSwapRequestSites(ctx, tx, spec.WorkspaceID, spec.InstanceID, domain, peerID, peerDomain); err != nil {
return err
}
_, err := integration.RecordLifecycleRequest(ctx, tx, spec)
_, err := integration.RecordLifecycleRequestInTx(ctx, tx, spec)
return err
})
}
+15 -13
View File
@@ -219,15 +219,17 @@ type HoldRequest struct {
// other caller of the package checks errors.Is(err, ErrNoUsageRow) and goes
// on.
func Hold(ctx context.Context, tx *sql.Tx, req HoldRequest) (held bool, err error) {
return hold(ctx, tx, req)
return hold(ctx, tx, tx, req)
}
func hold(ctx context.Context, tx entitlements.DBTX, req HoldRequest) (held bool, err error) {
// hold is Hold with the lock and its read on tx and every other statement on
// q, a handle a test wraps to make one of them fail.
func hold(ctx context.Context, tx *sql.Tx, q entitlements.DBTX, req HoldRequest) (held bool, err error) {
row, err := lockRequest(ctx, tx, req)
if err != nil {
return false, err
}
siteQ := fwmod.New(tx)
siteQ := fwmod.New(q)
alreadyActive, err := activeSite(ctx, siteQ, req.InstanceID, row.Operation)
if err != nil {
return false, err
@@ -237,35 +239,35 @@ func hold(ctx context.Context, tx entitlements.DBTX, req HoldRequest) (held bool
return false, fmt.Errorf("read hold of request %s: %w", row.RequestID, err)
}
if existing {
return recountHeld(ctx, tx, row.WorkspaceID)
return recountHeld(ctx, q, row.WorkspaceID)
}
poolID, err := primaryPool(ctx, tx, row.WorkspaceID)
poolID, err := primaryPool(ctx, q, row.WorkspaceID)
if err != nil {
return false, err
}
limit, found, err := Limit(ctx, entitlements.New(tx), poolID)
limit, found, err := Limit(ctx, entitlements.New(q), poolID)
if err != nil {
return false, err
}
if !found {
// No entitlement is a full limit; the pool's figure, when it has a
// row, is still written.
if _, err := recount(ctx, tx, poolID); err != nil && !errors.Is(err, ErrNoUsageRow) {
if _, err := recount(ctx, q, poolID); err != nil && !errors.Is(err, ErrNoUsageRow) {
return false, err
}
return false, nil
}
slots, err := lockAndCount(ctx, tx, poolID)
slots, err := lockAndCount(ctx, q, poolID)
if err != nil {
return false, err
}
if !alreadyActive && slots.Total() >= limit {
return false, slots.write(ctx, tx, poolID)
return false, slots.write(ctx, q, poolID)
}
if err := siteQ.InsertSlotHold(ctx, row.RequestID); err != nil {
return false, fmt.Errorf("hold a slot for request %s: %w", row.RequestID, err)
}
if _, err := recount(ctx, tx, poolID); err != nil {
if _, err := recount(ctx, q, poolID); err != nil {
return false, err
}
return true, nil
@@ -287,11 +289,11 @@ func recountHeld(ctx context.Context, tx entitlements.DBTX, workspaceID string)
// lockRequest takes the request's instance lock and reads the request, which
// must still be accepted.
func lockRequest(ctx context.Context, tx entitlements.DBTX, req HoldRequest) (integration.LifecycleRequest, error) {
q := integration.New(tx)
if err := q.LockLifecycleInstance(ctx, req.InstanceID); err != nil {
func lockRequest(ctx context.Context, tx *sql.Tx, req HoldRequest) (integration.LifecycleRequest, error) {
if err := db.Lock(ctx, tx, db.Lifecycle, req.InstanceID); err != nil {
return integration.LifecycleRequest{}, err
}
q := integration.New(tx)
row, err := q.GetLifecycleRequestByAddress(ctx, integration.GetLifecycleRequestByAddressParams{
InstanceID: req.InstanceID, Generation: req.Generation,
})
@@ -355,7 +355,7 @@ func TestHoldReturnsAFailedLimitRead(t *testing.T) {
var held bool
err := db.InTx(f.ctx, f.db, db.ReadCommitted, func(tx *sql.Tx) error {
var err error
held, err = usage.HoldOn(f.ctx, failingLimitRead{tx}, usage.HoldRequest{InstanceID: instance, Generation: gen})
held, err = usage.HoldOn(f.ctx, tx, failingLimitRead{tx}, usage.HoldRequest{InstanceID: instance, Generation: gen})
return err
})
if err == nil || held {
@@ -270,7 +270,7 @@ func holdIntakeDeletion(t *testing.T, env *siteEnv, target, domain string, repla
t.Fatal(err)
}
holderPID := int(dbtest.On(t, holder).Int("SELECT pg_backend_pid()"))
if err := integration.New(holder).LockLifecycleInstance(env.ctx, target); err != nil {
if err := db.Lock(env.ctx, holder, db.Lifecycle, target); err != nil {
t.Fatal(err)
}
dbtest.On(t, holder).Exec("DELETE FROM fedwiki.sites WHERE site_id = $1", target)
@@ -615,7 +615,7 @@ func (h *FedWikiPartialsHandler) swapActiveSite(w http.ResponseWriter, r *http.R
return err
}
answer, err = integration.RecordLifecycleRequest(ctx, tx, newRequestSpec(conn, ws, act.session.PersonID,
answer, err = integration.RecordLifecycleRequestInTx(ctx, tx, newRequestSpec(conn, ws, act.session.PersonID,
integration.LifecycleSetStatus, act.addr, workflowID,
requestParams{Domain: act.site.Domain, Status: "active", SwapParkSiteID: parked.SiteID}))
if err != nil {
@@ -787,7 +787,7 @@ func (h *FedWikiPartialsHandler) siteActionConflict(w http.ResponseWriter) {
}
// withWorkspaceTx runs fn inside a transaction that holds a per-workspace
// Postgres advisory lock (pg_advisory_xact_lock, released on commit or
// Postgres advisory lock (the fedwiki-workspace family's, released on commit or
// rollback), serializing a workspace's active-site rotations so two tabs
// cannot interleave the cooldown check-and-stamp or the park-target
// selection (audit #51). fn receives the transaction and a querier on it;
@@ -798,10 +798,7 @@ func (h *FedWikiPartialsHandler) withWorkspaceTx(ctx context.Context, workspaceI
return errors.New("no database for the workspace lock")
}
return db.InTx(ctx, h.Database, db.ReadCommitted, func(tx *sql.Tx) error {
// hashtext maps the workspace UUID text onto the bigint the advisory-lock
// API takes; an occasional hash collision only serializes two unrelated
// workspaces briefly, which is harmless.
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, workspaceID); err != nil {
if err := db.Lock(ctx, tx, db.FedWikiWorkspace, workspaceID); err != nil {
return err
}
return fn(ctx, tx, fwmod.New(tx))
@@ -1030,7 +1027,7 @@ func (h *FedWikiPartialsHandler) recordParkRequest(ctx context.Context, tx *sql.
return requestAddress{}, integration.LifecycleAnswer{}, err
}
park := requestAddress{Instance: instance, Generation: gens[instance]}
answer, err := integration.RecordLifecycleRequest(ctx, tx, newRequestSpec(h.connectionID(), act.session.WorkspaceID, act.session.PersonID,
answer, err := integration.RecordLifecycleRequestInTx(ctx, tx, newRequestSpec(h.connectionID(), act.session.WorkspaceID, act.session.PersonID,
integration.LifecycleSetStatus, park, workflowID,
requestParams{Domain: domain, Status: "readonly", SwapActivateSiteID: act.site.SiteID}))
return park, answer, err
@@ -0,0 +1,50 @@
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package web
import (
"context"
"database/sql"
"testing"
"time"
"github.com/google/uuid"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
fwmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/store"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
)
// TestWorkspaceTxTakesTheWorkspaceLock holds that a rotation's transaction
// waits for the lock of its own workspace and for no other: a side
// transaction holds one workspace's lock, the rotation of that workspace
// waits, and the rotation of another workspace goes through.
func TestWorkspaceTxTakesTheWorkspaceLock(t *testing.T) {
database := dbtest.Open(t)
h := &FedWikiPartialsHandler{Database: database}
ctx := context.Background()
held, other := uuid.NewString(), uuid.NewString()
noop := func(context.Context, *sql.Tx, fwmod.Querier) error { return nil }
q := dbtest.Tx(t, database)
sidePID := int(q.Int(`SELECT pg_backend_pid()`))
if err := db.Lock(ctx, q.Tx(), db.FedWikiWorkspace, held); err != nil {
t.Fatalf("take the workspace lock on the side: %v", err)
}
rotation := make(chan error, 1)
go func() { rotation <- h.withWorkspaceTx(ctx, held, noop) }()
dbtest.AwaitLockWait(t, database, dbtest.Backend{BlockedBy: sidePID})
otherCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
if err := h.withWorkspaceTx(otherCtx, other, noop); err != nil {
t.Errorf("a rotation of another workspace waited for workspace %s's lock: %v", held, err)
}
if err := q.Tx().Rollback(); err != nil {
t.Fatalf("release the workspace lock: %v", err)
}
if err := <-rotation; err != nil {
t.Errorf("the rotation returned %v", err)
}
}
@@ -14,7 +14,7 @@ import (
"testing"
"time"
"git.coopcloud.tech/wiki-cafe/member-console/internal/integration"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
fwmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/store"
"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest"
)
@@ -78,7 +78,7 @@ func retentionCancelledWait(env *reqEnv, site fwmod.Site, row purgeFailureCase)
env.t.Helper()
farm := observePurgeFarm(env, nil)
side := env.beginSide(env.lockWaitContext())
if err := integration.New(side.tx).LockLifecycleInstance(side.ctx, site.SiteID); err != nil {
if err := db.Lock(side.ctx, side.tx, db.Lifecycle, site.SiteID); err != nil {
env.t.Fatal(err)
}
connection, pid := purgeConnection(env.t, env.db)
@@ -13,7 +13,6 @@ import (
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
"git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements"
"git.coopcloud.tech/wiki-cafe/member-console/internal/integration"
fwmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/store"
)
@@ -148,7 +147,7 @@ var errRetentionFarmDelete = errors.New("retention farm delete")
func (a *Activities) purgeArchivedSite(ctx context.Context, database db.Beginner, candidate fwmod.Site, cutoff time.Time) (bool, error) {
deleted := false
err := db.InTx(ctx, database, db.ReadCommitted, func(tx *sql.Tx) error {
if err := integration.New(tx).LockLifecycleInstance(ctx, candidate.SiteID); err != nil {
if err := db.Lock(ctx, tx, db.Lifecycle, candidate.SiteID); err != nil {
return err
}
siteQ := fwmod.New(tx)
@@ -247,7 +247,7 @@ func finishAllWith(ctx context.Context, database db.Beginner, outs []integration
func lockOutcomes(ctx context.Context, tx *sql.Tx, outs []integration.LifecycleOutcome) (held []integration.LifecycleRequest, finished int, err error) {
q := integration.New(tx)
for _, out := range outs {
if err := q.LockLifecycleInstance(ctx, out.InstanceID); err != nil {
if err := db.Lock(ctx, tx, db.Lifecycle, out.InstanceID); err != nil {
return nil, 0, err
}
row, err := q.GetLifecycleRequestByAddress(ctx, integration.GetLifecycleRequestByAddressParams{
@@ -172,20 +172,19 @@ func noEnd() {}
// outbox row, permanently "sync in progress" and unretryable, since the
// dead-letter retry only matches outbox rows that were created; (2) two
// concurrent syncs both cleared the check-then-act guards in Sync and
// enqueued duplicate create_stripe_product entries. The per-product
// advisory lock (the same idiom as fulfillment's reconcile) serializes
// them, and re-reading the mappings under the lock makes the loser a no-op
// instead of a duplicate. Only an enqueue commits. From the begin on,
// enqueue returns end, which rolls the transaction back; it does not roll
// back itself, so a no-op or a failure keeps the lock until the caller
// calls end (Sync).
// enqueued duplicate create_stripe_product entries. The stripe-product-sync
// family's lock (internal/db/locks.go) serializes them, and re-reading the
// mappings under the lock makes the loser a no-op instead of a duplicate.
// Only an enqueue commits. From the begin on, enqueue returns end, which rolls
// the transaction back; it does not roll back itself, so a no-op or a failure
// keeps the lock until the caller calls end (Sync).
func enqueue(ctx context.Context, d DB, productID string, product billing.Product, price billing.Price, keyMode string) (Outcome, func(), error) {
tx, err := db.Begin(ctx, d, db.Plain)
if err != nil {
return 0, noEnd, &Error{Stage: StageBegin, Err: err}
}
end := func() { tx.Rollback() }
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, productID); err != nil {
if err := db.Lock(ctx, tx, db.ProductSync, productID); err != nil {
return 0, end, &Error{Stage: StageLock, Err: err}
}
outcome, err := enqueueLocked(ctx, tx, productID, product, price, keyMode)
@@ -415,10 +415,10 @@ func (r *invoiceReconcile) converge(ctx context.Context, customerID string) (Inv
}
defer tx.Rollback()
// The form reconcile.go uses for subscriptions. A concurrent reconcile
// of the same invoice (two events, or an event and the sweep) waits
// here and then finds what this one wrote.
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, r.id); err != nil {
// The stripe-invoice family's lock (internal/db/locks.go). A concurrent
// reconcile of the same invoice (two events, or an event and the sweep)
// waits here and then finds what this one wrote.
if err := db.Lock(ctx, tx, db.Invoice, r.id); err != nil {
return "", fmt.Errorf("stripe invoice %s (%s): advisory lock: %w", r.id, r.trig.Ref, err)
}
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
GetCustomerMappingByStripeCustomerID
AssignNextInvoiceNumber
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
GetCustomerMappingByStripeCustomerID
AssignNextInvoiceNumber
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
GetCustomerMappingByStripeCustomerID
AssignNextInvoiceNumber
@@ -1,7 +1,7 @@
== statements
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
GetCustomerMappingByStripeCustomerID
AssignNextInvoiceNumber
@@ -7,7 +7,7 @@ stripe GET /v1/invoices/in_<invoice:A>
stripe GET /v1/charges limit=100&payment_intent=pi_A
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetSubscriptionMappingByStripeID
GetInvoiceMappingByStripeID
ConvergeInvoice
@@ -22,7 +22,7 @@ MarkInvoiceMappingReadBack
stripe GET /v1/invoices/in_<invoice:B>
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
@@ -31,7 +31,7 @@ commit
stripe GET /v1/invoices/in_<invoice:C>
GetCustomerMappingByStripeCustomerID
begin
SELECT pg_advisory_xact_lock(hashtext($1)::bigint)
LockInvoice
GetInvoiceMappingByStripeID
ConvergeInvoice
ConvergeInvoiceMapping
+10
View File
@@ -59,6 +59,9 @@ type Config struct {
// spec-purpose-placeholder rule; "-" disables it.
// Default: openspec/specs.
SpecsDir string
// LockRoot is the repository root the lock rules walk (ruleLockSites,
// ruleLockFamilies); "-" disables them. Default: the current directory.
LockRoot string
}
// Violation is a single rule failure with source location.
@@ -99,6 +102,9 @@ func Run(cfg Config) (*Result, error) {
if cfg.SpecsDir == "" {
cfg.SpecsDir = "openspec/specs"
}
if cfg.LockRoot == "" {
cfg.LockRoot = "."
}
routes, err := extractRoutes(cfg.ServerDir)
if err != nil {
@@ -140,6 +146,10 @@ func Run(cfg Config) (*Result, error) {
res.Violations = append(res.Violations, ruleSpecPurposePlaceholder(cfg.SpecsDir)...)
}
res.Violations = append(res.Violations, ruleHTMX4Scripts(filepath.Join(filepath.Dir(cfg.TemplateDir), "static"))...)
if cfg.LockRoot != "-" {
res.Violations = append(res.Violations, ruleLockSites(cfg.LockRoot)...)
res.Violations = append(res.Violations, ruleLockFamilies(cfg.LockRoot)...)
}
sort.Slice(res.Violations, func(i, j int) bool {
if res.Violations[i].File != res.Violations[j].File {
+437
View File
@@ -0,0 +1,437 @@
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package lint
import (
"errors"
"fmt"
"go/ast"
"go/parser"
"go/token"
"io/fs"
"os"
"path/filepath"
"regexp"
"slices"
"strconv"
"strings"
"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
)
// The lock helpers (docs/database-locks.md): every advisory lock is taken in
// advisoryHelper, every pool row lock through the two queries poolHelper
// calls.
const (
advisoryHelper = "internal/db/locks.go"
poolHelper = "internal/entitlements/tx.go"
locksDoc = "docs/database-locks.md"
)
// advisoryCallRe is an advisory lock function named in SQL text, in any case:
// SQL function names are not case-sensitive.
var advisoryCallRe = regexp.MustCompile(`(?i)\bpg_(?:try_)?advisory_`)
// rowLockRe is a row-locking clause.
var rowLockRe = regexp.MustCompile(`(?i)\bFOR\s+(?:UPDATE|NO\s+KEY\s+UPDATE|SHARE|KEY\s+SHARE)\b`)
// poolLockQueries names, by query file, the one block of each that may lock
// a core.resource_pools row.
var poolLockQueries = map[string]string{
"internal/entitlements/queries/resource_pools.sql": "LockResourcePool",
"internal/entitlements/queries/grants.sql": "LockGrantPool",
}
// lockSeeds are the seeds that take the materialization rendezvous.
var lockSeeds = map[string]bool{
"test/mockshot-seed.sql": true,
"test/seed/discourse/seed-discourse.sh": true,
}
// advisoryMigrations are the two migrations that name the rendezvous, by file:
// a new migration's wrapper function would otherwise let any query lock
// outside the helpers.
var advisoryMigrations = map[string]bool{
"internal/db/migrations/00018_entitlement_set_changes.sql": true,
"internal/db/migrations/00040_rendezvous_family_key.sql": true,
}
// advisoryAllowed reports whether the file at rel may name an advisory lock
// function: the helper; the test kit's probe and its own wait tests, which
// lock integers no family builds; this rule and its tests, which name the
// rendezvous statement a seed may run and write fixtures that take locks;
// the two migrations of advisoryMigrations; and internal/db's tests, which
// run migrations at older versions against the key each version checks,
// every rendezvous use of which core.assert_rendezvous checks.
func advisoryAllowed(rel string) bool {
switch rel {
case advisoryHelper, "internal/testkit/dbtest/locks.go", "internal/testkit/dbtest/wait_test.go",
"internal/lint/locks.go", "internal/lint/locks_test.go":
return true
}
return advisoryMigrations[rel] ||
(strings.HasPrefix(rel, "internal/db/") && strings.HasSuffix(rel, "_test.go"))
}
// outsidePoolRule reports whether rel is outside the pool row rule: a test,
// the test kit, generated query code, a migration or a shell script.
func outsidePoolRule(rel string) bool {
return strings.HasSuffix(rel, "_test.go") || strings.HasPrefix(rel, "internal/testkit/") ||
strings.HasSuffix(rel, ".sql.go") || strings.Contains(rel, "/migrations/") || strings.HasSuffix(rel, ".sh")
}
// relPath is path relative to root, with forward slashes.
func relPath(root, path string) string {
rel, err := filepath.Rel(root, path)
if err != nil {
return filepath.ToSlash(path)
}
return filepath.ToSlash(rel)
}
// lockFiles reads every .go, .sql and .sh file under cmd/, scripts/,
// internal/ and test/ of root, and the .go files of root itself. It does not
// enter a testdata directory, which holds fixtures and, under test/, the
// stack's own data, some of it owned by another user, so readTree, which
// enters every directory, cannot serve here.
func lockFiles(root string) ([]sourceFile, []Violation) {
var files []sourceFile
read := func(path string) error {
switch filepath.Ext(path) {
case ".go", ".sql", ".sh":
default:
return nil
}
src, err := os.ReadFile(path)
if errors.Is(err, fs.ErrNotExist) {
return nil
}
if err != nil {
return err
}
files = append(files, sourceFile{path: path, src: src})
return nil
}
if entries, err := os.ReadDir(root); err == nil {
for _, e := range entries {
if !e.IsDir() && filepath.Ext(e.Name()) == ".go" {
if err := read(filepath.Join(root, e.Name())); err != nil {
return files, []Violation{unreadableFile(root, err)}
}
}
}
}
for _, dir := range []string{"cmd", "scripts", "internal", "test"} {
base := filepath.Join(root, dir)
if _, err := os.Stat(base); errors.Is(err, fs.ErrNotExist) {
continue
}
if err := walkLockDir(base, read); err != nil {
return files, []Violation{unreadableFile(base, err)}
}
}
return files, nil
}
// walkLockDir calls read with every file under base and does not enter a
// testdata directory or a directory whose name begins with a dot, such as the
// agent runner's disposable copy of the tree, scripts/agent-runner/.code.
func walkLockDir(base string, read func(path string) error) error {
return filepath.WalkDir(base, func(path string, d fs.DirEntry, err error) error {
if err != nil {
return err
}
if d.IsDir() {
if d.Name() == "testdata" || strings.HasPrefix(d.Name(), ".") {
return filepath.SkipDir
}
return nil
}
return read(path)
})
}
// ruleLockSites reports every lock under the directories lockFiles reads
// taken outside its helper.
func ruleLockSites(root string) []Violation {
files, v := lockFiles(root)
key, _ := db.Materialization.Key() // a family with no id parts always builds its key
rendezvous := "pg_advisory_xact_lock(hashtextextended('" + key + "', 0))"
for _, f := range files {
rel := relPath(root, f.path)
v = append(v, advisorySites(f, rel, rendezvous)...)
if !outsidePoolRule(rel) {
v = append(v, poolSites(f, rel)...)
}
}
return v
}
// advisorySites reports each Go string literal of f that names an advisory
// lock function where rel may not, and each line of the code of a .sql or
// .sh file that does (its comments removed, so a comment never matches); in
// a seed, each one but the rendezvous's, and the rendezvous anywhere but as
// the first statement after `BEGIN;`. A chain of literals joined by + is
// one text, so a function name in a literal of its own is found; Go
// comments are not strings and never match. A Go file the parser refuses is
// reported unread here and not again by poolSites.
func advisorySites(f sourceFile, rel, rendezvous string) []Violation {
if advisoryAllowed(rel) {
return nil
}
const message = "an advisory lock outside internal/db/locks.go: take it with db.Lock or db.LockShared and its family (docs/database-locks.md)"
if strings.HasSuffix(rel, ".go") {
strs, v := literalChains(f, "advisory locks")
for _, s := range strs {
if advisoryCallRe.MatchString(s.text) {
v = append(v, Violation{File: f.path, Line: s.line, Rule: "advisory-lock-site", Message: message})
}
}
return v
}
shell := strings.HasSuffix(rel, ".sh")
var v []Violation
prev := "" // the last line of code before this one, trimmed
heredoc := "" // the delimiter of the shell heredoc the line is in, if any
for i, line := range strings.Split(string(f.src), "\n") {
inDoc := heredoc != ""
code := ""
if inDoc && strings.TrimSpace(line) == heredoc {
heredoc = ""
} else {
code = stripComment(line, shell && !inDoc)
}
if shell && !inDoc {
if m := heredocRe.FindStringSubmatch(line); m != nil {
heredoc = m[1]
}
}
if strings.HasPrefix(strings.TrimSpace(code), "--") {
code = "" // a SQL comment line, in a shell file's heredoc too
}
calls := advisoryCallRe.FindAllString(code, -1)
switch {
case len(calls) == 0:
case lockSeeds[rel] && len(calls) == 1 && strings.Contains(code, rendezvous) && strings.EqualFold(prev, "BEGIN;"):
case lockSeeds[rel] && len(calls) == 1 && strings.Contains(code, rendezvous):
v = append(v, Violation{File: f.path, Line: i + 1, Rule: "advisory-lock-site",
Message: "a seed takes the materialization rendezvous other than as the first statement after BEGIN (docs/database-locks.md)"})
case lockSeeds[rel]:
v = append(v, Violation{File: f.path, Line: i + 1, Rule: "advisory-lock-site",
Message: "a seed takes an advisory lock other than the materialization rendezvous, " + rendezvous + " (internal/db/locks.go)"})
default:
v = append(v, Violation{File: f.path, Line: i + 1, Rule: "advisory-lock-site", Message: message})
}
if t := strings.TrimSpace(code); t != "" {
prev = t
}
}
return v
}
// heredocRe is the opening of a shell heredoc, whose body is SQL text.
var heredocRe = regexp.MustCompile(`<<-?\s*['"]?([A-Za-z_]\w*)`)
// stripComment returns line without its comment: in shell code, from a #
// that starts the line or follows a space or tab; in SQL text, from a --.
// A marker inside quotes is not a comment.
func stripComment(line string, shell bool) string {
var quote byte
for i := 0; i < len(line); i++ {
c := line[i]
switch {
case quote != 0:
if c == quote {
quote = 0
}
case c == '\'' || (shell && c == '"'):
quote = c
case !shell && strings.HasPrefix(line[i:], "--"):
return line[:i]
case shell && c == '#' && (i == 0 || line[i-1] == ' ' || line[i-1] == '\t'):
return line[:i]
}
}
return line
}
// poolSites reports, in a query file, each block that locks a pool row and
// is not the file's allowed one; in a Go file, each string literal, or chain
// of literals joined by +, that locks a pool row and, outside poolHelper,
// each call or method value of a pool lock query.
func poolSites(f sourceFile, rel string) []Violation {
if strings.HasSuffix(rel, ".sql") {
if !strings.Contains(rel, "/queries/") {
return nil
}
return poolQueryBlocks(f, rel)
}
v := poolStrings(f)
if rel == poolHelper {
return v
}
return append(v, poolCalls(f)...)
}
// poolCalls reports each selector of LockResourcePool or LockGrantPool in a
// Go file, called or taken as a method value; a comment or a string naming
// one never matches.
func poolCalls(f sourceFile) []Violation {
fset := token.NewFileSet()
file, err := parser.ParseFile(fset, f.path, f.src, parser.SkipObjectResolution)
if err != nil {
return nil // advisorySites reports a file the parser refuses
}
var v []Violation
ast.Inspect(file, func(n ast.Node) bool {
if s, ok := n.(*ast.SelectorExpr); ok && (s.Sel.Name == "LockResourcePool" || s.Sel.Name == "LockGrantPool") {
v = append(v, Violation{File: f.path, Line: fset.Position(s.Sel.Pos()).Line, Rule: "pool-lock-site",
Message: "LockResourcePool or LockGrantPool called outside internal/entitlements/tx.go: lock pools through LockPools or LockPoolOfGrant (docs/go-conventions.md)"})
}
return true
})
return v
}
// locksPool reports whether SQL text, its -- comments removed, names
// resource_pools and takes a row lock.
func locksPool(text string) bool {
var b strings.Builder
for _, line := range strings.Split(text, "\n") {
code, _, _ := strings.Cut(line, "--")
b.WriteString(code + "\n")
}
s := b.String()
return strings.Contains(s, "resource_pools") && rowLockRe.MatchString(s)
}
// poolQueryBlocks reports each "-- name:" block of a query file that locks
// a pool row, unless it is the file's allowed block (poolLockQueries); text
// before the first block is a block with no name.
func poolQueryBlocks(f sourceFile, rel string) []Violation {
var v []Violation
name, start := "", 1
var block strings.Builder
flush := func() {
if locksPool(block.String()) && (name == "" || name != poolLockQueries[rel]) {
v = append(v, Violation{File: f.path, Line: start, Rule: "pool-lock-site",
Message: fmt.Sprintf("query %s locks a core.resource_pools row: lock pools through LockPools or LockPoolOfGrant (docs/go-conventions.md)", name)})
}
block.Reset()
}
for i, line := range strings.Split(string(f.src), "\n") {
if rest, ok := strings.CutPrefix(strings.TrimSpace(line), "-- name: "); ok {
flush()
name, _, _ = strings.Cut(rest, " ")
start = i + 1
continue
}
block.WriteString(line + "\n")
}
flush()
return v
}
// goString is a string literal of a Go file, or a chain of literals joined
// by +, with the line of its first literal.
type goString struct {
line int
text string
}
// literalChains returns every goString of f. A file the parser refuses comes back
// as a violation naming what has not been checked.
func literalChains(f sourceFile, checked string) ([]goString, []Violation) {
fset := token.NewFileSet()
file, err := parser.ParseFile(fset, f.path, f.src, parser.SkipObjectResolution)
if err != nil {
return nil, []Violation{{File: f.path, Rule: "unreadable-file", Message: "the lint cannot parse this file, so it has not checked its " + checked + ": " + err.Error()}}
}
var out []goString
ast.Inspect(file, func(n ast.Node) bool {
e, ok := n.(ast.Expr)
if !ok {
return true
}
text, ok := literalChain(e)
if !ok {
return true
}
out = append(out, goString{line: fset.Position(e.Pos()).Line, text: text})
return false
})
return out, nil
}
// literalChain is the text of a string literal, or of a chain of string
// literals joined by +; it is false when any leaf of the expression is not a
// string literal.
func literalChain(e ast.Expr) (string, bool) {
switch e := e.(type) {
case *ast.BasicLit:
if e.Kind != token.STRING {
return "", false
}
s, err := strconv.Unquote(e.Value)
return s, err == nil
case *ast.BinaryExpr:
if e.Op != token.ADD {
return "", false
}
l, ok := literalChain(e.X)
if !ok {
return "", false
}
r, ok := literalChain(e.Y)
return l + r, ok
}
return "", false
}
// poolStrings reports each goString of a Go file that locks a pool row.
func poolStrings(f sourceFile) []Violation {
strs, _ := literalChains(f, "pool row locks") // advisorySites reports a file the parser refuses
var v []Violation
for _, s := range strs {
if locksPool(s.text) {
v = append(v, Violation{File: f.path, Line: s.line, Rule: "pool-lock-site",
Message: "a core.resource_pools row lock outside LockResourcePool and LockGrantPool: lock pools through LockPools or LockPoolOfGrant (docs/go-conventions.md)"})
}
}
return v
}
// ruleLockFamilies reports the family table of docs/database-locks.md when
// its names differ from db.Families, in a name or in order. The table is
// the one under "## Advisory key families"; each row's first cell is a
// family's name in backticks.
func ruleLockFamilies(root string) []Violation {
path := filepath.Join(root, locksDoc)
src, err := os.ReadFile(path)
if err != nil {
return []Violation{unreadableFile(path, err)}
}
var listed []string
heading, in := 0, false
for i, line := range strings.Split(string(src), "\n") {
switch {
case strings.HasPrefix(line, "## Advisory key families"):
in, heading = true, i+1
case in && strings.HasPrefix(line, "## "):
in = false
case in && strings.HasPrefix(line, "| `"):
name, _, _ := strings.Cut(strings.TrimPrefix(line, "| `"), "`")
listed = append(listed, name)
}
}
want := make([]string, 0, len(db.Families))
for _, f := range db.Families {
want = append(want, f.Name())
}
if slices.Equal(listed, want) {
return nil
}
return []Violation{{File: path, Line: heading, Rule: "lock-families",
Message: fmt.Sprintf("the advisory key families table lists %v, want %v, the order of db.Families (internal/db/locks.go)", listed, want)}}
}

Some files were not shown because too many files have changed in this diff Show More