@qrvey/dataset-lifecycle

Authoritative dataset state machine for Qrvey: atomic guarded transitions, finalization claims (fencing tokens), a TTL self-heal, and standardized lifecycle events. It is the single owner of a dataset's dataloadingStatus so that concurrent writers (qrvey_qollect handlers, Data Router, cancel requests) can never corrupt the state (DP-3816).

• Node.js ≥ 20
• Peer dependency: @qrvey/data-persistence. The library owns all query building and its own qv_dataset schema, and connects through data-persistence for you. In the turnkey path you pass nothing about persistence — it uses the standard MULTIPLATFORM_PG_CONNECTION_STRING environment. You can still supply your own pool, CrudService, or a fully custom adapter.

Installation

npm install @qrvey/dataset-lifecycle @qrvey/data-persistence

Concepts

• Transition (CAS): every status change goes through transition(), which runs UPDATE ... WHERE dataloadingStatus IN (:from) AND jobId = :ownerJobId. If the row was already changed, it returns { applied: false, reason } — no throw.
• Finalization claim: a fencing token (finalizingJobId + finalizingScope + finalizingClaimTime) that guarantees only one handler does destructive finalization work at a time. Self-heals via a TTL (staleClaimMs, default 5 min). The claim is held per (jobId, scope), not per jobId — Data Router can deliver more than one terminal signal carrying the same jobId, so the finalizer identity has to be part of the key. See Claim scoping.
• Reservation field (optional): a caller-specific "busy" flag (e.g. qollect's __DR.isJobRunning) that is cleared atomically in the same write when a transition reaches a terminal state — no separate unreserve write.
• FSM: transitions are validated against TRANSITION_TABLE; illegal ones throw InvalidTransitionError.

Initialization

Call init() once at service startup, before any other method.

Turnkey (recommended)

Pass only business configuration — the library builds its own CrudService against qv_dataset and connects per-query via MULTIPLATFORM_PG_CONNECTION_STRING. No pool, adapter or mapper needed.

import { DatasetLifecycle } from '@qrvey/dataset-lifecycle';

DatasetLifecycle.init({
    logger,                          // optional (info/warn/error)
    eventPublisher,                  // optional
    reservationField: '__DR.isJobRunning', // optional — cleared on terminal transitions
    staleClaimMs: 5 * 60 * 1000,     // optional — finalization claim TTL (default 5 min)
    loadGuards: {                    // optional — atomic "not busy" rule on → LOADING
        '__DR.isJobRunning': [null, false],
        '__DR.isJLORunning': [null, false],
        queuedLoadToken: null,
    },
});

Reuse an existing pool (optional optimization)

Lifecycle writes are low-frequency, so per-query connections are usually fine. If you prefer to reuse your service's shared pool (no per-query connect), pass it:

import { DatasetLifecycle } from '@qrvey/dataset-lifecycle';

DatasetLifecycle.init({ pool: mySharedPgPool, /* ...same options */ });

Bring your own schema (recommended when payloads write many columns)

If your finalizers persist extra dataset columns atomically via the transition payload (e.g. metadataId, indexName, rowCount), pass your full CrudSchema so the library can write them. The built-in minimal schema only knows the lifecycle columns.

import { DatasetLifecycle } from '@qrvey/dataset-lifecycle';
import { TableSchema } from './config/schemas/datasetSchema';

DatasetLifecycle.init({ crudSchema: TableSchema, /* ...same options */ });

Bring your own CrudService

import { CrudService } from '@qrvey/data-persistence';
import { DatasetLifecycle } from '@qrvey/dataset-lifecycle';

DatasetLifecycle.init({ crudService: new CrudService(MyDatasetSchema, pool) });

With a custom persistence adapter

If your model does not expose the CrudService interface (method names / nested-path expression syntax differ), implement the small PersistenceAdapter contract directly:

const adapter = {
    async updateByExpression(table, updateExpression, conditionExpression) {
        // translate the plain { field: value } shapes to your model's API;
        // return true if a row was updated, false on 0-rows-affected (CAS miss)
    },
    async getByKeys(table, keys) {
        // return the full record for { userId, datasetId } or null
    },
};
DatasetLifecycle.init({ persistence: adapter, reservationField: '__DR.isJobRunning' });

The reservationField value may be a nested path (e.g. __DR.isJobRunning); the injected adapter is responsible for translating it. On any transition to a non-active state (not LOADING/CANCELLING) the library sets it to false in the same atomic write that finalizes the status.

Config options (init)

Provide at most one persistence source (persistence > crudService > pool). If none is given, the library builds its own CrudService and connects via MULTIPLATFORM_PG_CONNECTION_STRING.

| Option | Required | Default | Description | | ------------------ | -------- | ----------------------- | ----------------------------------------------------------------- | | crudSchema | ❌ | built-in minimal | A CrudSchema — needed if payloads write non-lifecycle columns | | pool | ❌ | env connection string | A pg Pool to reuse; otherwise per-query connections are used | | crudService | ❌ | built-in | A ready @qrvey/data-persistence CrudService | | persistence | ❌ | built-in | A fully custom PersistenceAdapter (escape hatch) | | eventPublisher | ❌ | none | Emits DatasetStatusChanged on applied transitions | | logger | ❌ | no-op | { info, warn, error } | | schema | ❌ | data_sources_datasets | DB schema for the built-in CrudService | | tableName | ❌ | qv_dataset | Table (PostgreSQL alias) for the built-in CrudService | | reservationField | ❌ | none | Field cleared to false on terminal transitions (unreserve) | | staleClaimMs | ❌ | 300000 (5 min) | Finalization claim TTL before it can be stolen | | loadGuards | ❌ | none | CAS conditions auto-injected on → LOADING (atomic "not busy") |

State machine

NOT_AVAILABLE → LOADING
LOADING       → SUCCEEDED | FAILED | CANCELLING | CANCELED
CANCELLING    → CANCELED  | FAILED | SUCCEEDED
SUCCEEDED     → LOADING   | NOT_AVAILABLE
FAILED        → LOADING   | NOT_AVAILABLE
CANCELED      → LOADING   | NOT_AVAILABLE

CANCELLING → SUCCEEDED is intentionally allowed ("complete wins"): if a cancel arrives after the job already finished, Data Router emits Complete (and no TerminatedExit), so finalizing as SUCCEEDED keeps the dataset consistent with DR's job status and avoids a stuck state / orphaned index.

STATUSES aliases: IDLE = NOT_AVAILABLE, CANCELLED = CANCELED.

Usage

Transition (any status change)

import { DatasetLifecycle, STATUSES } from '@qrvey/dataset-lifecycle';

const result = await DatasetLifecycle.transition(identifier, {
    from: [STATUSES.LOADING, STATUSES.CANCELLING], // optional; omit to auto-read current
    to: STATUSES.SUCCEEDED,
    ownerJobId: 'job-abc',
    payload: { metadataId, rowCount },   // optional extra fields written atomically
    eventData: { recordsAdded: 100 },    // optional event payload
    guards: { '__DR.isJobRunning': [null, false] }, // optional extra CAS conditions
});

if (!result.applied) {
    // result.reason: 'STATE_MISMATCH' | 'NOT_OWNER' | 'GUARD_FAILED' | 'RECORD_NOT_FOUND'
    return { statusCode: 208 };
}

Finalization claim + transition (status handlers)

const scope = 'complete'; // 'complete' | 'cancel' | 'failure'

const claim = await DatasetLifecycle.claimFinalization(identifier, jobId, {
    scope,
});
if (!claim.claimed) {
    // claim.reason: 'HELD_BY_OTHER' | 'RECORD_NOT_FOUND'
    // claim.currentHolder / claim.currentScope say who holds it
    return { statusCode: 208 };
}
try {
    // ... destructive finalization (apply changes, delete versions, swap index) ...
    const res = await DatasetLifecycle.transition(identifier, {
        from: [STATUSES.LOADING, STATUSES.CANCELLING],
        to: STATUSES.SUCCEEDED,
        ownerJobId: jobId,
        payload: datasetStatus,
    });
    if (!res.applied) return { statusCode: 208 };
} finally {
    // owner-checked (jobId AND scope) + idempotent: no-op if the transition
    // already cleared the claim
    await DatasetLifecycle.releaseClaim(identifier, jobId, { scope });
}

A successful terminal transition() clears the claim (and reservationField) atomically, so releaseClaim in a finally is a safe no-op on the happy path and only matters when finalization returns early (208) or throws.

Claim scoping

scope names which finalizer is claiming. It matters because a single load can produce two terminal signals with the same jobId: Data Router writes Complete and notifies, a concurrent terminate whose read predated that write overwrites the row with TerminatedExit, and a later delivery notifies again. Both messages reach the status queue, which has competing consumers across containers.

Keyed on jobId alone, the "idempotent re-claim" branch would admit both the complete and the cancel handler: both would proceed, their destructive work would interleave, and only the closing CAS would arbitrate — by which point the complete handler may already have promoted the new index for the cancel handler to delete. That is the corruption DP-3816 was filed for.

Keyed on (jobId, scope):

| Arrival order | Outcome | | ------------------------ | --------------------------------------------------------------------------------- | | complete claims first | cancel gets HELD_BY_OTHER → 208, never runs. Dataset finalizes SUCCEEDED. | | cancel claims first | complete gets HELD_BY_OTHER → 208. Nothing was promoted, so CANCELED is clean. | | same finalizer redelivered| re-claim succeeds (that is the branch's purpose) and the CAS keeps it idempotent. |

Omit scope and the claim falls back to jobId-only keying (scope stored as NULL). That is only safe when exactly one finalizer can exist per job.

Busy checks

const { busy, reasons } = await DatasetLifecycle.checkBusy(identifier);
// reasons: 'DATALOADING_ACTIVE' | 'JOB_RUNNING' | 'JLO_RUNNING' | 'LOAD_QUEUED'
const isBusy = await DatasetLifecycle.isBusy(identifier); // boolean shortcut
const state = await DatasetLifecycle.getLifecycleState(identifier); // full snapshot or null

Ask "can I?" and query state (guarded, rule-owned)

The library owns the rules; callers state intent and get back a yes/no with all the reasons why not. canTransitionTo is a dry-run of the exact rules transition enforces, so the "ask" and the "do" never diverge.

await DatasetLifecycle.getStatus(id);   // 'LOADING' | 'SUCCEEDED' | ... | null
await DatasetLifecycle.isLoading(id);   // boolean

// "I'm about to inject data — can I move it to LOADING?"
const check = await DatasetLifecycle.canTransitionTo(id, 'LOADING');
// check = { allowed: false, reasons: ['JOB_RUNNING', 'LOAD_QUEUED'], currentStatus: 'SUCCEEDED' }
if (!check.allowed) {
    // tell the user exactly why: another job/JLO is running, a load is queued, etc.
}

Starting a load (→ LOADING) is blocked if the dataset is busy by any signal (dataloadingStatus active, JOB_RUNNING, JLO_RUNNING, LOAD_QUEUED). transition() returns the same structured reasons[] when its CAS is rejected.

When loadGuards is configured in init, transition(to: 'LOADING') enforces "not busy" atomically by auto-injecting those guards into the CAS WHERE (the caller never passes them — the library owns the rule). [null, false] is null-aware, so field IS NULL OR field = false is honoured:

DatasetLifecycle.init({
    persistence: adapter,
    loadGuards: {
        '__DR.isJobRunning': [null, false], // IS NULL OR = false
        '__DR.isJLORunning': [null, false],
        queuedLoadToken: null,              // IS NULL
    },
});
// transition(to: 'LOADING') now fails atomically (applied:false) if the dataset
// became busy between the check and the write — no TOCTOU window.

Blocking reasons (BlockReason): RECORD_NOT_FOUND, INVALID_TRANSITION, DATALOADING_ACTIVE, JOB_RUNNING, JLO_RUNNING, LOAD_QUEUED, NOT_OWNER, FINALIZATION_HELD.

Testing

import { MockDatasetLifecycle } from '@qrvey/dataset-lifecycle/testing';

const lifecycle = new MockDatasetLifecycle();
lifecycle.setInitialState(identifier, {
    dataloadingStatus: 'LOADING',
    jobId: 'job-1',
    finalizingJobId: null,
    finalizingScope: null,
    finalizingClaimTime: null,
});

const result = await lifecycle.transition(identifier, {
    to: 'SUCCEEDED',
    ownerJobId: 'job-1',
});
// lifecycle.transitionCalls / lifecycle.claimCalls record invocations.

The mock validates against the real TRANSITION_TABLE, so illegal transitions fail in tests exactly as they would in production.

Public exports

• DatasetLifecycle — singleton (init, transition, claimFinalization, releaseClaim, getLifecycleState, checkBusy, isBusy, canTransitionTo, getStatus, isLoading, reset)
• createPersistenceAdapter, STATUSES, TRANSITION_TABLE, isValidTransition, evaluateTransition, STALE_CLAIM_MS
• Errors: InvalidTransitionError, LifecycleNotInitializedError, ValidationError
• Types: DatasetStatus, LifecycleConfig, TransitionOptions, TransitionResult, ClaimResult, ReleaseResult, LifecycleState, BusyCheckResult, BusyReason, BlockReason, TransitionCheckResult, PersistenceAdapter, EventPublisher, Logger, DatasetIdentifier, DatasetLifecycleApi (shared runtime contract — implemented by both the singleton and MockDatasetLifecycle; use it to type DI slots)
• @qrvey/dataset-lifecycle/testing → MockDatasetLifecycle

Required DB columns

The dataset table needs three nullable columns (plus a partial index for sweeps). finalizingJobId holds a copy of qv_dataset."jobId", so it is declared with the same unbounded width rather than a narrower one:

ALTER TABLE data_sources_datasets.qv_dataset
    ADD COLUMN IF NOT EXISTS "finalizingJobId"     VARCHAR(255) DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS "finalizingScope"     VARCHAR(32)  DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS "finalizingClaimTime" TIMESTAMPTZ  DEFAULT NULL;

CREATE INDEX IF NOT EXISTS idx_qv_dataset_finalizing_claim
    ON data_sources_datasets.qv_dataset
       ("finalizingJobId", "finalizingScope", "finalizingClaimTime")
    WHERE "finalizingJobId" IS NOT NULL;