Skip to content

Add automatic data-attribute-to-Postgres sync#308

Open
longquanzheng wants to merge 3 commits into
mainfrom
long/pg-integ
Open

Add automatic data-attribute-to-Postgres sync#308
longquanzheng wants to merge 3 commits into
mainfrom
long/pg-integ

Conversation

@longquanzheng

Copy link
Copy Markdown
Contributor

What

Adds an opt-in, SDK-side feature to automatically mirror a workflow data attribute to a single cell (row + column) of a user-owned Postgres table:

  • Load on start — the data attribute is loaded from its mapped column when the workflow starts.
  • Sync on mutation — when the data attribute is mutated in a state's execute, the new value is written back to the column.

The iWF data attribute stays the source of truth; the DB cell is a mirror. The SDK is driver-agnostic (depends only on javax.sql.DataSource); the user owns the schema, credentials, and pooling.

API

DataAttributeDef.create(String.class, "status",
    DbAttributeSync.of(dataSource, "public.orders", "id",
        (workflowId, persistence) -> new CellLocation(workflowId, "status_col")));

How it works

  • On start, Client swaps the start state to a worker-internal load state that SELECTs each mapped column, sets the data attribute, then transitions to the real starting state (original input passed through).
  • When a mapped data attribute is mutated in execute, the worker reroutes the decision through a worker-internal sync state (carrying the captured original decision as DbSyncStatePayload) that UPDATEs the column, then replays the original decision verbatim — so any decision shape (single/multi-next, graceful/force complete/fail, conditionalClose) is preserved.
  • iWF persists the data attribute first, then the DB write happens, so a DB failure retries only the sync step and never re-runs user execute code.
  • Neither system state is registered in the Registry; both are handled specially in WorkerService by reserved stateId.

Security

  • Credentials never held by the SDK (user-supplied DataSource).
  • All values bound via PreparedStatement; identifiers validated against an allowlist and double-quoted.

Tests / CI

  • DbSyncTest runs end-to-end against a real Postgres (script/docker-compose.yml iwf-integ-postgres + script/postgres-init/iwf-integ-init.sql) and a live iWF server; verified locally (load + sync).
  • CI adds a Postgres readiness wait before the build.

Scope (v1)

  • Fixed-key data attributes only (not createByPrefix).
  • Sync on execute mutations only (not waitUntil).
  • Load fires for starts via the registered Client.
  • The mapped column stores the attribute's JSON — use text/jsonb.

Docs

longquanzheng and others added 3 commits July 7, 2026 22:31
Mirror a workflow data attribute to a cell of a user-owned Postgres table:
the iWF data attribute stays the source of truth; the DB cell is a mirror.

- DataAttributeDef.create(type, key, DbAttributeSync) declares the mapping
  (DataSource + table + pk column + a (workflowId, persistence) -> CellLocation locator)
- On start, Client swaps the start state to a worker-internal load state that
  SELECTs each mapped column, sets the data attribute, then transitions to the
  real starting state with the original input passed through.
- When a state's execute mutates a mapped data attribute, the worker reroutes the
  decision through a worker-internal sync state (carrying DbSyncStatePayload) that
  UPDATEs the column, then replays the original decision verbatim. iWF persists
  first, so a DB failure retries only the sync step, not user code.
- New: CellLocation, DbAttributeSync, PostgresDataAttributeSyncer (parameterized
  values + allowlisted/quoted identifiers, try-with-resources), DbSyncStatePayload.
- Neither system state is registered; both are handled specially in WorkerService.
- Integration test (DbSyncTest) against a real Postgres via docker-compose +
  init script (schema/table/seed); CI wired with a Postgres readiness wait.
- Docs: user guide + concise overview, linked from README.

Scope (v1): fixed-key data attributes only; sync on execute mutations only;
load fires for starts via the registered Client; column stores the attribute JSON.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…rver readiness

The pinned Elasticsearch 7.16.2 crashes on startup on cgroup-v2 hosts (current
GitHub runners) with a bundled-JDK NPE in DefaultSystemMemoryInfo/CgroupInfo, so
ES never starts, Temporal's frontend never becomes ready, and every integration
test fails on connectivity. ES 7.17.28 bundles a cgroup-v2-safe JDK and is
compatible with Temporal 1.25 (verified locally: ES cluster health green,
iWF server ready).

Also replace the fixed `sleep 30` before the build with an active wait that polls
the iWF server until it serves, so the build only starts once the full
Temporal + Elasticsearch stack is actually ready.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant