Skip to content

sync: port module changes from constructive-db #129

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pyramation merged 2 commits into from
Sep 9, 2026
Merged

sync: port module changes from constructive-db #129

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0fb9407
sync: port module changes from constructive-db
pyramation Sep 9, 2026
d848ff9
test(app-scope): seed principal_auth_module.private_schema_id now tha…
pyramation Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion packages/app-scope/__tests__/actor-entity-type.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,13 +178,14 @@ describe('app_scope.actor_entity', () => {
(
database_id,
schema_id,
private_schema_id,
principals_table_id,
principal_entities_table_id,
users_table_id,
sessions_table_id,
session_credentials_table_id
)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
VALUES ($1, $2, $2, $3, $4, $5, $6, $7)`,
[
PRINCIPAL_DATABASE_ID,
principalSchema.id,
Expand Down
33 changes: 33 additions & 0 deletions packages/database-jobs/__tests__/jobs.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -457,6 +457,39 @@ describe('scheduled jobs', () => {
});
});

it('run_scheduled_job re-runs a keyed job that is out of attempts instead of wedging on it', async () => {
// constructive-planning#2013: a per-minute schedule with max_attempts = 1
// whose first tick failed (fail_job clears locked_at, so the dead job read
// as "never been run") raised ALREADY_SCHEDULED on every later tick.
const scheduled = await pg.one(
`INSERT INTO app_jobs.scheduled_jobs (
database_id, task_identifier, schedule_info, key, max_attempts
) VALUES ($1, $2, $3, $4, 1)
RETURNING id, key`,
[database_id, 'dead_keyed_job', { rule: '0 * * * * *' }, 'dead_keyed_job']
);
const [first] = await pg.any(`SELECT * FROM app_jobs.run_scheduled_job($1)`, [scheduled.id]);

// A tick while the job is still waiting to run is covered by it.
await expect(
pg.any(`SELECT * FROM app_jobs.run_scheduled_job($1)`, [scheduled.id])
).rejects.toThrow('ALREADY_SCHEDULED');

// The worker claims it, fails it, and the attempt budget is gone.
await pg.any(`SELECT * FROM app_jobs.get_job('worker-1', ARRAY[$1::text])`, ['dead_keyed_job']);
await pg.any(`SELECT * FROM app_jobs.fail_job('worker-1', $1, $2)`, [
first.id,
'Function "dead_keyed_job" is not registered in function_definitions'
]);
const dead = await pg.one(`SELECT attempts, max_attempts, locked_at FROM app_jobs.jobs WHERE id = $1`, [first.id]);
expect(dead).toEqual({ attempts: 1, max_attempts: 1, locked_at: null });

// The next tick replaces it with a fresh attempt on the same row.
const [again] = await pg.any(`SELECT * FROM app_jobs.run_scheduled_job($1)`, [scheduled.id]);
expect(again.id).toBe(first.id);
expect({ attempts: again.attempts, last_error: again.last_error }).toEqual({ attempts: 0, last_error: null });
});

it('run_scheduled_job rejects a malformed entity pair under strict attribution', async () => {
const task_identifier = 'malformed_scheduled_job';
await pg.any(`BEGIN`);
Expand Down
5 changes: 5 additions & 0 deletions packages/database-jobs/deploy/schemas/app_jobs/procedures/run_scheduled_job.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,11 @@ BEGIN
app_jobs.jobs js
WHERE
js.id = sched.last_scheduled_id
-- a job out of attempts never runs again (get_job skips it), so it
-- covers nothing: fail_job cleared its locked_at, which otherwise reads
-- as "never been run" and wedges the schedule on a permanently failed
-- job. The keyed upsert below replaces it with a fresh attempt instead.
AND js.attempts < js.max_attempts

@tenki-reviewer tenki-reviewer Bot Sep 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 bug · high

Non-keyed schedule duplicates in-flight last-attempt job

The new probe condition AND js.attempts < js.max_attempts (packages/database-jobs/deploy/schemas/app_jobs/procedures/run_scheduled_job.sql:42) also excludes a job that is currently locked and running on its final attempt (attempts == max_attempts, locked_at set), even though such a job still covers the tick. For a schedule with sched.key IS NULL the keyed in-flight guard at lines 60-71 is skipped, so the function inserts a brand-new job while the previous tick is still executing, producing two concurrent executions of the same schedule. A non-keyed schedule with max_attempts = 1 whose worker is still processing when the next tick fires now enqueues a duplicate instead of raising ALREADY_SCHEDULED.

📋 Prompt for AI Agents

In packages/database-jobs/deploy/schemas/app_jobs/procedures/run_scheduled_job.sql around line 42, change the already-scheduled probe so that a locked (in-flight) job always raises ALREADY_SCHEDULED regardless of attempt budget. Replace AND js.attempts < js.max_attempts with AND (js.attempts < js.max_attempts OR js.locked_at IS NOT NULL). Rationale: the fix intended to let a permanently-dead keyed job (locked_at NULL, attempts >= max_attempts) be refreshed, but it also excludes a job currently running on its final attempt (locked_at NOT NULL, attempts == max_attempts), which still covers the tick. For non-keyed schedules the keyed in-flight guard at lines 60-71 is skipped, so this regression enqueues a duplicate concurrent job instead of raising ALREADY_SCHEDULED.

AND (js.locked_at IS NULL -- never been run
OR js.locked_at >= (NOW() - job_expiry)
-- still running within a safe interval
Expand Down
Binary file modified packages/database-jobs/sql/pgpm-database-jobs--0.44.0.bundle.tar.gz
View file
Open in desktop
Binary file not shown.
5 changes: 5 additions & 0 deletions packages/database-jobs/sql/pgpm-database-jobs--0.44.0.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -433,6 +433,11 @@ BEGIN
app_jobs.jobs js
WHERE
js.id = sched.last_scheduled_id
-- a job out of attempts never runs again (get_job skips it), so it
-- covers nothing: fail_job cleared its locked_at, which otherwise reads
-- as "never been run" and wedges the schedule on a permanently failed
-- job. The keyed upsert below replaces it with a fresh attempt instead.
AND js.attempts < js.max_attempts
AND (js.locked_at IS NULL -- never been run
OR js.locked_at >= (NOW() - job_expiry)
-- still running within a safe interval
Expand Down
33 changes: 29 additions & 4 deletions packages/function-resolution/__tests__/capabilities.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ describe('function-resolution capability resolution', () => {
required_buckets text[] NOT NULL DEFAULT '{}',
required_modules text[] NOT NULL DEFAULT '{}',
required_models text[] NOT NULL DEFAULT '{}',
integrations text[] NOT NULL DEFAULT '{}'
integrations text[] NOT NULL DEFAULT '{}',
required_capabilities jsonb
)`
);
await pg.query(
Expand Down Expand Up @@ -176,9 +177,10 @@ describe('function-resolution capability resolution', () => {
resource_installations_table_id, apps_table_id, buckets_table_id,
sites_web_config_table_id, sites_error_pages_table_id,
sites_app_links_table_id, sites_deep_links_table_id,
app_store_identities_table_id,
images_table_id, redirects_table_id,
app_store_identities_table_id, bindings_table_id, scope)
VALUES ($1, $2, $3, $3, $3, $3, $4, $3, $3, $3, $3, $3, $3, $5, $3, $3, $3, $3, $3, $3, $6, 'database')`,
bindings_table_id, scope)
VALUES ($1, $2, $3, $3, $3, $4, $3, $3, $3, $3, $3, $3, $5, $3, $3, $3, $3, $3, $3, $3, $6, 'database')`,
[
TENANT_DB,
catFunctions.schemaId,
Expand Down Expand Up @@ -272,8 +274,17 @@ describe('function-resolution capability resolution', () => {
[TENANT_DB]
);
ids.ambiguous = ambiguous.id;
const declaring = await pg.one(
`INSERT INTO cap_defs.function_definitions
(database_id, task_identifier, access_channels, required_buckets, required_capabilities)
VALUES ($1, 'signup:welcome', ARRAY['api'], ARRAY['exports'],
'[{"name":"email","version":"1.0.0"},{"name":"events","version":"1.0.0","declaration":{"stream":"signups"}}]'::jsonb)
RETURNING id`,
[TENANT_DB]
);
ids.declaring = declaring.id;

for (const id of [ids.exporter, ids.ambiguous]) {
for (const id of [ids.exporter, ids.ambiguous, ids.declaring]) {
await pg.query(
`INSERT INTO catalog_private.functions (id, owner_scope, owner_key, is_visible, database_id, task_identifier)
SELECT d.id, 'database', d.database_id, false, d.database_id, d.task_identifier
Expand Down Expand Up @@ -512,6 +523,20 @@ describe('function-resolution capability resolution', () => {
expect(bundle.apis['notifications_module'].api_id).toBe(ids.adminApi);
expect(bundle.models).toEqual(['gpt-4o']);
expect(bundle.payload).toEqual({ subject: 'monthly' });
// A definition that declared no capabilities echoes null, not [] — the
// runtime mounts the full platform set for it.
expect(bundle.capabilities).toBeNull();
});

it('resolve_capabilities(): echoes required_capabilities as declared', async () => {
const [{ bundle }] = await pg.any(
`SELECT function_resolution.resolve_capabilities($1, 'database', $1, $2, 'database', $1, '{}'::jsonb, 'api') AS bundle`,
[TENANT_DB, ids.declaring]
);
expect(bundle.capabilities).toEqual([
{ name: 'email', version: '1.0.0' },
{ name: 'events', version: '1.0.0', declaration: { stream: 'signups' } },
]);
});

it('resolve_capabilities(): an explicit binding overrides tag discovery', async () => {
Expand Down
2 changes: 1 addition & 1 deletion packages/function-resolution/__tests__/published-capability-catalog.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,7 +245,7 @@ describe('capability resolution against the published catalog planes', () => {
sites_error_pages_table_id,
sites_app_links_table_id, sites_deep_links_table_id,
images_table_id, redirects_table_id, scope)
VALUES ($1, $2, $3, $3, $3, $3, $4, $3, $3, $3, $3, $3, $3, $5, $6, $3, $3, $3, $3, $3, $3, 'database')`,
VALUES ($1, $2, $3, $3, $3, $4, $3, $3, $3, $3, $3, $3, $5, $3, $6, $3, $3, $3, $3, $3, $3, 'database')`,
[dbId, schemaId, bucketsTableId, apisTableId, bucketsTableId, bindingsTableId]
);
// Label kept for readability of the fixture rows above.
Expand Down
15 changes: 11 additions & 4 deletions ...ction-resolution/deploy/schemas/function_resolution/procedures/install_route_bindings.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,14 @@ BEGIN;
--
-- The bindings document is a JSON array whose every entry NAMES ITS TARGET KIND:
--
-- {"path": "/login", "target": "function", "task_identifier": "mantra:signin"}
-- {"path": "/login", "target": "function", "task_identifier": "mantra:signin", "anonymous": true}
-- {"path": "/app", "target": "service", "service_id": "<uuid>"}
--
-- An entry may declare `anonymous`, which is the route's half of the anonymous
-- contract: the URL answers callers carrying no identity. It opens nothing on
-- its own — the definition behind it must declare anonymous_callable too — and
-- defaults to false, so a document that says nothing installs closed routes.
--
-- The kind is never inferred from which key happens to be present, and an entry
-- carrying keys for two kinds is a malformed document rather than a precedence
-- question: guessing is how a deployment silently binds /app to the wrong plane.
Expand Down Expand Up @@ -124,6 +129,7 @@ DECLARE
entry_target text;
entry_task text;
entry_service uuid;
entry_anonymous boolean;
target_column text;
target_id uuid;
service_found boolean;
Expand Down Expand Up @@ -374,6 +380,7 @@ BEGIN
LOOP
entry_path := entry ->> 'path';
entry_target := entry ->> 'target';
entry_anonymous := coalesce((entry ->> 'anonymous')::boolean, false);

IF entry_target = 'function' THEN
entry_task := entry ->> 'task_identifier';
Expand Down Expand Up @@ -426,8 +433,8 @@ BEGIN

-- pgsql-lint-disable-next-line no-dynamic-sql -- write-only: insert into the routes plane named by app_scope.routing_tables; every value is a bound parameter
query := format(
'INSERT INTO %I.%I (%s%sdomain_id, path, %I)
SELECT %s%s$1, $2, $3
'INSERT INTO %I.%I (%s%sdomain_id, path, anonymous, %I)
SELECT %s%s$1, $2, $6, $3
WHERE NOT EXISTS (
SELECT 1 FROM %I.%I AS x
WHERE x.domain_id = $1 AND x.path = $2%s)',
Expand All @@ -447,7 +454,7 @@ BEGIN
);

EXECUTE query USING domain_id, entry_path, target_id, key_value,
install_route_bindings.site_id;
install_route_bindings.site_id, entry_anonymous;
GET DIAGNOSTICS inserted = ROW_COUNT;

IF inserted > 0 THEN
Expand Down
3 changes: 3 additions & 0 deletions ...unction-resolution/deploy/schemas/function_resolution/procedures/resolve_capabilities.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,6 +265,9 @@ BEGIN
'configs', coalesce(v_definition->'required_configs', '[]'::jsonb),
'integrations', coalesce(v_definition->'integrations', '[]'::jsonb),
'access_channels', coalesce(v_definition->'access_channels', '[]'::jsonb),
-- Not coalesced: NULL means the handler declared nothing and gets the
-- full platform set, an empty array means it declared none.
'capabilities', v_definition->'required_capabilities',
'payload', function_resolution.resolve_payload_refs(
resolve_capabilities.database_id,
resolve_capabilities.scope,
Expand Down
Binary file modified packages/function-resolution/sql/pgpm-function-resolution--0.44.2.bundle.tar.gz
View file
Open in desktop
Binary file not shown.
11 changes: 8 additions & 3 deletions packages/function-resolution/sql/pgpm-function-resolution--0.44.2.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1566,6 +1566,9 @@ BEGIN
'configs', coalesce(v_definition->'required_configs', '[]'::jsonb),
'integrations', coalesce(v_definition->'integrations', '[]'::jsonb),
'access_channels', coalesce(v_definition->'access_channels', '[]'::jsonb),
-- Not coalesced: NULL means the handler declared nothing and gets the
-- full platform set, an empty array means it declared none.
'capabilities', v_definition->'required_capabilities',
'payload', function_resolution.resolve_payload_refs(
resolve_capabilities.database_id,
resolve_capabilities.scope,
Expand Down Expand Up @@ -1695,6 +1698,7 @@ DECLARE
entry_target text;
entry_task text;
entry_service uuid;
entry_anonymous boolean;
target_column text;
target_id uuid;
service_found boolean;
Expand Down Expand Up @@ -1945,6 +1949,7 @@ BEGIN
LOOP
entry_path := entry ->> 'path';
entry_target := entry ->> 'target';
entry_anonymous := coalesce((entry ->> 'anonymous')::boolean, false);

IF entry_target = 'function' THEN
entry_task := entry ->> 'task_identifier';
Expand Down Expand Up @@ -1997,8 +2002,8 @@ BEGIN

-- pgsql-lint-disable-next-line no-dynamic-sql -- write-only: insert into the routes plane named by app_scope.routing_tables; every value is a bound parameter
query := format(
'INSERT INTO %I.%I (%s%sdomain_id, path, %I)
SELECT %s%s$1, $2, $3
'INSERT INTO %I.%I (%s%sdomain_id, path, anonymous, %I)
SELECT %s%s$1, $2, $6, $3
WHERE NOT EXISTS (
SELECT 1 FROM %I.%I AS x
WHERE x.domain_id = $1 AND x.path = $2%s)',
Expand All @@ -2018,7 +2023,7 @@ BEGIN
);

EXECUTE query USING domain_id, entry_path, target_id, key_value,
install_route_bindings.site_id;
install_route_bindings.site_id, entry_anonymous;
GET DIAGNOSTICS inserted = ROW_COUNT;

IF inserted > 0 THEN
Expand Down
Loading
Loading