Add automatic data-attribute-to-Postgres sync

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>

Author: longquanzheng

.github/workflows/ci-integ-test.yml

@@ -20,6 +20,15 @@ jobs:
       - uses: gradle/gradle-build-action@v2
         with:
           gradle-version: 7.5
+      - name: "Wait for integ Postgres to be ready"
+        run: |
+          for i in $(seq 1 30); do
+            if docker exec iwf-integ-postgres pg_isready -U iwf -d iwf_integ; then
+              echo "postgres is ready"; exit 0
+            fi
+            echo "waiting for postgres... ($i)"; sleep 2
+          done
+          echo "postgres did not become ready in time" >&2; exit 1
       - run: git submodule update --init --recursive && sleep 30 && gradle build
       - name: Dump docker logs
         if: always()

README.md

@@ -53,6 +53,11 @@ A workflow can contain any number of WorkflowStates.
 
 See more in https://github.com/indeedeng/iwf#what-is-iwf
 
+## Features
+
+* [Data Attribute → Postgres sync](./docs/data-attribute-db-sync-overview.md) — automatically mirror a
+  data attribute to a cell of a user-owned Postgres table (load on start, sync on mutation).
+
 ## How to build & run
 
 ### Using IntelliJ

build.gradle

@@ -75,6 +75,10 @@ dependencies {
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation "javax.validation:validation-api:2.0.1.Final"
 
+    // Postgres JDBC driver for the data-attribute DB-sync integration test (test-only; the core SDK
+    // stays driver-agnostic and depends only on javax.sql.DataSource).
+    testImplementation 'org.postgresql:postgresql:42.7.4'
+
 
     // immutable
     compileOnly 'org.immutables:value:2.9.2'

docs/data-attribute-db-sync-overview.md

@@ -0,0 +1,36 @@
+# Data Attribute → Postgres sync (overview)
+
+Automatically mirror a workflow **data attribute** to a cell (row + column) of a user-owned Postgres
+table. The iWF data attribute stays the source of truth; the database cell is a mirror.
+
+- **Load on start** — when the workflow starts, the data attribute is loaded from its mapped column.
+- **Sync on mutation** — whenever the data attribute is mutated in a state's `execute` method, the new
+  value is written back to the same column.
+
+```java
+DataSource ds = /* your configured, pooled DataSource */;
+
+DataAttributeDef.create(
+    String.class,
+    "status",
+    DbAttributeSync.of(ds, "public.orders", "id",
+        (workflowId, persistence) -> new CellLocation(workflowId, "status_col")));
+```
+
+## How it works (3 lines)
+
+1. `Client.startWorkflow` starts at a load system state that `SELECT`s each mapped column and sets the
+   data attribute, then transitions to your real starting state (input passes through).
+2. When a state's `execute` mutates a mapped data attribute, the worker reroutes the decision through a
+   sync system state that `UPDATE`s the column, then continues to your original decision.
+3. iWF persists the data attribute first, then the DB write happens — so a DB failure retries only the
+   sync step, never your `execute`. Neither system state is registered; both are worker-internal.
+
+## Key limitations (v1)
+
+- Fixed-key data attributes only (not `createByPrefix`).
+- Sync fires on `execute` mutations only (not `waitUntil`).
+- Load fires for workflows started through the registered `Client`.
+- The mapped column stores the data attribute's JSON — use a `text`/`jsonb` column.
+
+See the detailed guide: [data-attribute-db-sync.md](./data-attribute-db-sync.md).

docs/data-attribute-db-sync.md

@@ -0,0 +1,164 @@
+# Data Attribute → Postgres sync (user guide)
+
+This feature lets you automatically mirror an iWF **data attribute** to a single cell (a row + column)
+of an arbitrary, **user-owned** Postgres table:
+
+- **Load on start**: when a workflow starts, the data attribute's initial value is loaded from the
+  mapped database column.
+- **Sync on mutation**: whenever the data attribute is mutated inside a state's `execute` method, the
+  new value is written back to the same column.
+
+It is entirely an **SDK-side** feature — iWF's server is not involved in your database, and you own the
+database schema, credentials, and connection pooling.
+
+## Mental model
+
+The iWF data attribute is the **source of truth**; the Postgres cell is a **mirror**.
+
+- On start, the mirror seeds the data attribute (load).
+- During execution, the data attribute drives the mirror (sync).
+- iWF persists the data attribute first, then the database is updated. If the database write fails, only
+  the (internal) sync step is retried — your `execute` code is **not** re-run.
+
+## Prerequisites
+
+1. A Postgres table you own, with:
+   - a **primary key column** (text/uuid/numeric — the PK value is bound as a string parameter), and
+   - a **data column** of type `text` or `jsonb` (the SDK stores the data attribute's JSON there).
+2. A configured `javax.sql.DataSource`. You own it — the SDK never stores credentials and never creates
+   connections other than via `dataSource.getConnection()`. Use any driver/pool (HikariCP,
+   `PGSimpleDataSource`, etc.).
+
+Example table:
+
+```sql
+CREATE TABLE orders (
+    id         TEXT PRIMARY KEY,
+    status_col TEXT
+);
+INSERT INTO orders (id, status_col) VALUES ('order-123', '"NEW"');
+```
+
+## Setup
+
+### 1. Build a DataSource
+
+```java
+import org.postgresql.ds.PGSimpleDataSource;
+import javax.sql.DataSource;
+
+PGSimpleDataSource ds = new PGSimpleDataSource();
+ds.setUrl("jdbc:postgresql://localhost:5432/mydb");
+ds.setUser("app");
+ds.setPassword(System.getenv("DB_PASSWORD")); // keep credentials out of code
+// (in production prefer a pooled DataSource, e.g. HikariCP)
+```
+
+### 2. Declare the data attribute with a DB-sync mapping
+
+Use the `DataAttributeDef.create(type, key, DbAttributeSync)` overload in your workflow's
+`getPersistenceSchema()`:
+
+```java
+import io.iworkflow.core.persistence.*;
+
+public class OrderWorkflow implements ObjectWorkflow {
+    public static final String STATUS = "status";
+
+    @Override
+    public List<PersistenceFieldDef> getPersistenceSchema() {
+        return Arrays.asList(
+            DataAttributeDef.create(
+                String.class,
+                STATUS,
+                DbAttributeSync.of(
+                    /* dataSource   */ MyDataSources.ORDERS,
+                    /* table        */ "orders",          // optionally schema-qualified: "public.orders"
+                    /* pk column    */ "id",
+                    /* locator      */ (workflowId, persistence) ->
+                        new CellLocation(workflowId, "status_col"))));
+    }
+
+    @Override
+    public List<StateDef> getWorkflowStates() {
+        return Arrays.asList(StateDef.startingState(new DecideState()));
+    }
+}
+```
+
+**The locator** `(String workflowId, Persistence persistence) -> CellLocation` returns which cell to
+load-from / sync-to:
+
+- `CellLocation.getPkValue()` — the primary-key **value** identifying the row (bound as a parameter).
+- `CellLocation.getColumnName()` — the **data column** name.
+
+The table name and the primary-key column name are fixed in `DbAttributeSync.of(...)`. The locator can
+compute the PK value from the `workflowId` and/or from the current `persistence` (other data attributes
+already set at that point).
+
+### 3. Use the data attribute normally
+
+```java
+public class DecideState implements WorkflowState<String> {
+    @Override public Class<String> getInputType() { return String.class; }
+
+    @Override
+    public StateDecision execute(Context ctx, String in, CommandResults r,
+                                 Persistence persistence, Communication comm) {
+        String status = persistence.getDataAttribute(OrderWorkflow.STATUS, String.class); // loaded from DB
+        // ... business logic ...
+        persistence.setDataAttribute(OrderWorkflow.STATUS, "PROCESSED"); // mirrored back to DB
+        return StateDecision.gracefulCompleteWorkflow(status);
+    }
+}
+```
+
+Start the workflow as usual with the registered `Client` — the load happens automatically:
+
+```java
+client.startWorkflow(OrderWorkflow.class, "order-123", 3600, "start");
+```
+
+## Behavior / semantics
+
+| Aspect | Behavior |
+|---|---|
+| When load fires | Once, at workflow start (via the registered `Client`). |
+| When sync fires | After a state's `execute` mutates a mapped data attribute. |
+| Ordering | iWF persists the data attribute first; then the DB `UPDATE` runs. |
+| Retry / idempotency | The DB write is a single-row idempotent `UPDATE`; safe to retry. On failure only the internal sync step retries — your `execute` is not re-run. |
+| Stored value | The data attribute's `ObjectEncoder` JSON (same bytes iWF stores). Use a `text`/`jsonb` column. |
+| Multiple synced attributes | Each maps to its own cell; all load at start and each syncs when mutated. |
+
+## Column & type guidance
+
+The mapped column stores the data attribute's JSON. For a `String` attribute of value `NEW`, the column
+holds the JSON string `"NEW"`. For complex objects it holds the serialized JSON object. This round-trips
+any type uniformly, so use `text` or `jsonb`. Mapping to a native scalar column (e.g. an `integer`
+column) is **not** supported in v1.
+
+## Limitations (v1)
+
+- **Fixed-key data attributes only** — `DataAttributeDef.createByPrefix(...)` is not supported for DB
+  sync (a `WorkflowDefinitionException` is thrown at registration if you try).
+- **`execute` mutations only** — a data attribute mutated in `waitUntil` (and not in `execute`) is not
+  synced. Mutate DB-synced data attributes in `execute`.
+- **Started via the registered `Client`** — starts issued directly against the server API / another SDK
+  with an explicit start stateId bypass the load step.
+
+## Troubleshooting
+
+| Symptom | Likely cause |
+|---|---|
+| Data attribute is `null` after start | Row/PK/column mismatch — the locator's PK value or column name doesn't match an existing row/column, or the column is null. |
+| Column not updated after a state | The attribute was mutated in `waitUntil` (not `execute`), or the mutated key isn't the DB-mapped one. |
+| `unsafe SQL identifier` error | The table / PK / data-column name isn't a plain identifier (`^[A-Za-z_][A-Za-z0-9_]*$`, optionally schema-qualified). Rename or quote-safe your identifiers. |
+| Connection / pool errors | Check the `DataSource` config, network access from the **worker** process to Postgres, and pool sizing. |
+| `WorkflowDefinitionException: DB sync is not supported for prefix data attributes` | You used `createByPrefix` with a `DbAttributeSync`; use a fixed key. |
+
+## Security notes
+
+- **Credentials** are never held by the SDK — you supply a configured `DataSource`.
+- **Values** (primary-key value and the data value) are always bound via `PreparedStatement` parameters.
+- **Identifiers** (table, PK column, data column) cannot be bound; they are validated against a strict
+  allowlist and double-quoted, so a malformed identifier is rejected rather than injected.

script/docker-compose.yml

@@ -66,6 +66,20 @@ services:
     volumes:
      - ./docker-compose-init.sh:/etc/temporal/init.sh
     entrypoint: sh -c "/etc/temporal/init.sh"
+  iwf-integ-postgres:
+    container_name: iwf-integ-postgres
+    image: postgres:${POSTGRESQL_VERSION}
+    environment:
+      POSTGRES_DB: iwf_integ
+      POSTGRES_USER: iwf
+      POSTGRES_PASSWORD: iwf
+    ports:
+      # host port 5455 (5432 is often occupied by other local Postgres containers); container is 5432
+      - 5455:5432
+    volumes:
+      - ./postgres-init:/docker-entrypoint-initdb.d
+    networks:
+      - temporal-network
   iwf-server:
     container_name: iwf-server
     image: iworkflowio/iwf-server:latest #NOTE: you can change to version tag like v1.0.0-RC6

script/postgres-init/iwf-integ-init.sql

@@ -0,0 +1,14 @@
+-- Initializes the schema/table and the initial dataset used by the data-attribute DB-sync
+-- integration test (io.iworkflow.integ.DbSyncTest). Postgres runs any file in
+-- /docker-entrypoint-initdb.d/ once on first container start, against POSTGRES_DB (iwf_integ).
+
+CREATE TABLE IF NOT EXISTS iwf_integ_data (
+    pk       TEXT PRIMARY KEY,
+    data_col TEXT
+);
+
+-- The value is stored as JSON text (the SDK mirrors the data attribute's ObjectEncoder JSON), so a
+-- String data attribute of value seeded-value is stored as the JSON string "seeded-value".
+INSERT INTO iwf_integ_data (pk, data_col)
+VALUES ('db-sync-test-key', '"seeded-value"')
+ON CONFLICT (pk) DO UPDATE SET data_col = EXCLUDED.data_col;

src/main/java/io/iworkflow/core/Client.java

@@ -175,6 +175,14 @@ public String startWorkflow(
             if (stateOptions != null) {
                 unregisterWorkflowOptions.startStateOptions(stateOptions);
             }
+
+            // If the workflow has DB-synced data attributes, start at the load system state instead of
+            // the real starting state. The load state reads the mapped columns and then transitions to
+            // the real starting state, passing the original input (sent as stateInput) through.
+            if (!registry.getDbAttributeSyncs(wfType).isEmpty()) {
+                startStateId = WorkerService.LOAD_DATA_ATTRIBUTES_FROM_DB_STATE_ID;
+                unregisterWorkflowOptions.startStateOptions(new WorkflowStateOptions().skipWaitUntil(true));
+            }
         }
 
         final PersistenceOptions schemaOptions = registry.getPersistenceOptions(wfType);

src/main/java/io/iworkflow/core/Registry.java

@@ -3,13 +3,15 @@
 import io.iworkflow.core.communication.InternalChannelDef;
 import io.iworkflow.core.communication.SignalChannelDef;
 import io.iworkflow.core.persistence.DataAttributeDef;
+import io.iworkflow.core.persistence.DbAttributeSync;
 import io.iworkflow.core.persistence.PersistenceFieldDef;
 import io.iworkflow.core.persistence.PersistenceOptions;
 import io.iworkflow.core.persistence.SearchAttributeDef;
 import io.iworkflow.gen.models.SearchAttributeValueType;
 
 import java.lang.reflect.Method;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -27,6 +29,8 @@ public class Registry {
     private final Map<String, TypeStore> signalTypeStore = new HashMap<>();
     private final Map<String, TypeStore> internalChannelTypeStore = new HashMap<>();
     private final Map<String, TypeStore> dataAttributeTypeStore = new HashMap<>();
+    // (workflow type) -> (data attribute key -> DB sync mapping)
+    private final Map<String, Map<String, DbAttributeSync>> dbAttributeSyncStore = new HashMap<>();
 
     private final Map<String, Map<String, SearchAttributeValueType>> searchAttributeTypeStore = new HashMap<>();
 
@@ -153,6 +157,17 @@ private void registerWorkflowDataAttributes(final ObjectWorkflow wf) {
 
         for (final DataAttributeDef dataAttributeField : fields) {
             typeStore.addToStore(dataAttributeField);
+
+            final DbAttributeSync dbSync = dataAttributeField.getDbSync();
+            if (dbSync != null) {
+                if (Boolean.TRUE.equals(dataAttributeField.isPrefix())) {
+                    throw new WorkflowDefinitionException(
+                            "DB sync is not supported for prefix data attributes: " + dataAttributeField.getKey());
+                }
+                dbAttributeSyncStore
+                        .computeIfAbsent(workflowType, s -> new HashMap<>())
+                        .put(dataAttributeField.getKey(), dbSync);
+            }
         }
     }
 
@@ -250,6 +265,14 @@ public TypeStore getDataAttributeTypeStore(final String workflowType) {
         return dataAttributeTypeStore.get(workflowType);
     }
 
+    /**
+     * @return the data-attribute-key -> DB sync mapping for the workflow type, or an empty map when
+     * the workflow has no DB-synced data attributes.
+     */
+    public Map<String, DbAttributeSync> getDbAttributeSyncs(final String workflowType) {
+        return dbAttributeSyncStore.getOrDefault(workflowType, Collections.emptyMap());
+    }
+
     public Map<String, SearchAttributeValueType> getSearchAttributeKeyToTypeMap(final String workflowType) {
         return searchAttributeTypeStore.get(workflowType);
     }