Attribute Store sync

View as Markdown

server-zd can sync Attributes to PostgreSQL. See External Attribute Sync for what it is and how to enable it, and Startup contracts for the table contract. A Flow marks an Attribute with syncToAttributeStore and lists a configured store name in its FlowConfig. server-zd then writes the Attribute to that store's tables.

Behavior

Only type: postgres is supported. Any other store type fails validation, so startup fails.
Step, RPC, and SetAttributes commits never wait for the table write.
Rows normally appear within a second of the commit. Treat the table as eventually consistent.
While a store is unavailable, Flows keep running. When the store recovers, server-zd writes the backlog with the latest values.
Entries still pending syncRetryPolicy.totalDuration after they were first marked are dropped. Later writes of the same Attributes sync normally.
After time travel, server-zd rewrites every row the run synced with the state after the time travel.
Each database needs a fencing table.
A natural close waits for synced Attributes. A Step decision that completes or fails the Flow, exhausted Step retries, or the Flow timeout stops the Steps at once. The run then stays RUNNING until every pending write is written or dropped. WaitForFlow returns only then. Entries written before the wait ended are in the table. A dropped entry does not guarantee the row's value, because earlier writes of the same run may have committed. If dexzd_attribute_sync_dropped_entries_total increases, compare the affected rows with the Flow's Attributes and write the Attributes again.
While a close waits, SetAttributes, PublishToChannel, and RPCs that are neither transactional nor locking are accepted. Their synced writes extend the wait, and they start no Step. Transactional or locking RPCs, WaitForStepCompletion, WaitForAttribute, and DeleteChannelMessage calls made during the wait fail with FAILED_PRECONDITION.
A natural close also waits for every transactional or locking RPC admitted before it, even without an Attribute Store. Such an RPC still commits, and its synced writes join the wait. The close is recorded once those RPCs commit, fail, or time out.
syncRetryPolicy.totalDuration limits how long each pending write is retried, not the whole close. During a PostgreSQL outage a close waits until its pending writes are dropped. Every synced write accepted while the close waits starts its own retry window and extends the wait.
StopFlow and Terminate never wait. StopFlow closes the run as soon as it commits, even while a close waits, and the pending writes keep syncing after the close. A later SetAttributes on that run returns NOT_FOUND. From server-zd v0.1.5, it succeeds only when every write erases a currently stored hard-delete Attribute with null. This exception applies only before the run is purged. It deletes the stored objects without reopening the run. The run's Flow history records the erase after the close. Write what a Flow needs before stopping it.
Pending writes for a store that is no longer configured are dropped at once and counted in dexzd_attribute_sync_dropped_entries_total.

Configuration

attributeStore:
  stores:
    orders:
      type: postgres
      dsn: ""
      tableName: public.orders
      rowKey: order-{order_id}
      columnNaming: snake_case
      attributeMaps:
        Items:
          tableName: public.order_items
          instanceKeyColumn: item_key
      fencingTableName: dexzd_attribute_sync_fencing
  schemaSyncInterval: 1m
  syncBatchSize: 100
  syncAttemptTimeout: 30s
  syncRetryPolicy:
    initialInterval: 100ms
    backoffCoefficient: 2.0
    maximumInterval: 30s
    maximumAttempts: 0
    totalDuration: 1h

An explicit zero for a duration, syncBatchSize, or a retry policy field means its default. The example configuration shows the section with comments.

Key	Default	Notes
stores	Empty	Maps each store name that a FlowConfig may list to a PostgreSQL destination. Empty disables sync and rejects every store name that a FlowConfig lists.
stores.NAME.type	None	Must be postgres.
stores.NAME.dsn	Empty	PostgreSQL connection string. When it is empty, server-zd reads an environment variable instead. Logs redact it.
stores.NAME.tableName	None	Main table, as table or schema.table.
stores.NAME.rowKey, columnNaming, and attributeMaps	Empty	See Row keys and column naming and AttributeMap tables.
stores.NAME.fencingTableName	dexzd_attribute_sync_fencing	Exists only in server-zd. See Fencing table.
schemaSyncInterval	1m	Time between table schema refreshes, with 90–110% jitter. A write to an unknown column refreshes the schema at once.
syncBatchSize	100	Most Attributes and AttributeMap instances of one run that one transaction writes. A run with more pending values uses several transactions.
syncAttemptTimeout	30s	Upper bound of one write transaction. See Write timeout.
syncRetryPolicy.initialInterval	100ms	Delay after the first failed write.
syncRetryPolicy.backoffCoefficient	2	Multiplies the delay after each failed write. Must be positive and finite.
syncRetryPolicy.maximumInterval	30s	Longest delay between attempts. Must not be less than initialInterval.
syncRetryPolicy.maximumAttempts	0, unlimited	Consecutive failed writes of one run's entries, after which the entries are dropped. The count lives in memory, so it starts again after a restart.
syncRetryPolicy.totalDuration	1h	Entries still pending this long after they were first marked are dropped.

Leave dsn empty to keep the connection string out of the configuration file. server-zd then reads DEX_ZD_ATTRIBUTE_STORE_DSN_ followed by the store name in upper case. Every character other than an ASCII letter or digit becomes an underscore. For the store orders, the variable is DEX_ZD_ATTRIBUTE_STORE_DSN_ORDERS. A store without a DSN in either place fails validation.

Fencing table

Every write transaction raises a fencing token row to the node's epoch. A node that lost its lease therefore cannot overwrite newer values.

Each database needs this table, named by fencingTableName. One table serves every store and every cluster in the database. server-zd never creates or alters tables. Create the table as the table owner before startup. Then grant SELECT, INSERT, and UPDATE on it to the role that server-zd connects as. Replace dex with that role.

CREATE TABLE dexzd_attribute_sync_fencing (
  store_name text NOT NULL,
  cluster_id text NOT NULL,
  scope text NOT NULL,
  fencing_token bigint NOT NULL,
  updated_at timestamptz NOT NULL,
  PRIMARY KEY (store_name, cluster_id, scope)
);
GRANT SELECT, INSERT, UPDATE ON dexzd_attribute_sync_fencing TO dex;

You may delete the rows of a permanently stopped cluster. Never delete the rows of a running cluster.

Startup connects to every store and checks its tables. Any failure fails startup, including an unreachable store or a missing fencing table. For a missing fencing table, the error includes the CREATE TABLE statement. Startup does not check grants. Without them, the fencing table looks missing, or every write fails with a permission error.

Write timeout

A write transaction lasts at most syncAttemptTimeout or a quarter of lease.duration, whichever is smaller. A quarter of lease.duration is also the lease margin. A write starts only while more than the transaction timeout plus the margin of lease validity remains. The PostgreSQL statement_timeout equals the transaction timeout. A COMMIT sent just before the deadline therefore still lands before the lease ends.

The timeout plus the margin is at most half of lease.duration. That is less than the two thirds of validity left when a lease renewal starts, so writes rarely wait for a renewal. With the default 10-second lease, a transaction lasts at most 2.5 seconds. With a 60-second lease, it lasts at most 15 seconds.

Retries and dropped entries

Failed writes are retried with the syncRetryPolicy backoff until totalDuration. The failure decides what backs off:

Store. A connection failure, or another failure of the whole store, backs off every run of that store.
Run. A failure caused by the written values or by a conflict with another session backs off only that run. Other Flows keep syncing. Examples are a constraint violation, a value too long for its column, a deadlock, and a serialization failure.
Timeout. A transaction that exceeds the transaction timeout, for example while it waits for a row lock, also backs off only that run.

After a timeout, the run's retries use half as many statements per transaction, down to one. After a timeout, one statement writes all of the run's plain Attributes in one transaction, so the main row is not partly updated. Each AttributeMap instance is one statement, so a timed-out run's instances can land in different transactions. Each transaction that commits is acknowledged, so a retry writes only the rest.

When three runs of a store time out with no commit in between, the whole store backs off. The retry after a store backoff starts with other runs than the one that failed. A newer run of the same Flow ID waits for the older run, so rows keep run order.

Entries still pending syncRetryPolicy.totalDuration after they were first marked are dropped with an error log. Later writes of the same Attributes sync normally. dexzd_attribute_sync_dropped_entries_total counts dropped entries. It also counts entries that the store rejected with a non-retryable error.

Monitoring

Metric	Meaning
dexzd_attribute_sync_backlog_entries	Entries waiting to be written, as of the last scan
dexzd_attribute_sync_oldest_pending_age_seconds	Age of the oldest pending entry's retry window, 0 when none
dexzd_attribute_sync_write_failures_total with failure_kind	Failed write transactions. failure_kind is retryable or non_retryable. Non-retryable failures drop their entries.
dexzd_attribute_sync_dropped_entries_total	Entries dropped after the retry policy was exhausted or the store rejected them
dexzd_attribute_sync_fenced_writes_total	Writes rejected because a newer owner raised the fencing token

# Dropped Attribute Sync entries
increase(dexzd_attribute_sync_dropped_entries_total[10m]) > 0

# Attribute Sync entries pending for more than 5 minutes
dexzd_attribute_sync_oldest_pending_age_seconds > 300

Startup contracts

The server connects, pings, and validates every destination during startup. One failure fails startup and closes already-opened clients.

Without rowKey, each table must have exactly one character-compatible primary key. Dex discovers that key and stores the Flow ID in it. With rowKey, the primary key must be exactly the columns the template names, and each must be a character column.

Every other required SQL column must have a default or generated value. This allows a partial insert to create a row. Attribute names match the remaining columns exactly, or by their snake_case form under columnNaming.

Row keys and column naming

A store accepts two optional settings. Without them, it writes each Flow ID to its single primary-key column.

rowKey splits each Flow ID into primary-key columns. Use it when a business ID is part of the Flow ID and you query by that ID, or when several Flow types describe one entity. A template contains literal text and placeholders:

{column} stores the captured text in that primary-key column.
{*} matches text without storing it.
Each placeholder captures up to the next literal text. The last placeholder captures the rest of the Flow ID.
Literal text must separate two placeholders, and no capture may be empty.

For example, order-{customer_id}-{order_id} splits order-c1-o-7 into customer_id c1 and order_id o-7. A Flow ID that does not match the template skips its whole batch with an error log, because a retry cannot change the result. Writes to a key column are filtered like Flow ID writes.

Several stores may name the same table. Each upsert sets only the columns of its own batch, so two Flow types can write different columns of one row. For example, a store with customer-{customer_id} and another with onboarding-{customer_id}-{*} both write the row of one customer.

columnNaming: snake_case writes each Attribute to the snake_case form of its name. Dex inserts an underscore before an upper-case letter that follows a lower-case letter or a digit, or that ends an acronym before a lower-case letter. Then it lower-cases every letter. DeploymentURL becomes deployment_url, URLPath becomes url_path, and S3Bucket becomes s3_bucket. When two Attribute names in one batch map to the same column, Dex filters both. The default writes each Attribute to the column with exactly its name.

AttributeMap tables

attributeMaps writes each instance of an AttributeMap as one row of its own table, so SQL can query map instances such as the line items of an order or the members of a group. Each key is an AttributeMap name:

tableName selects the instance table, as table or schema.table.
instanceKeyColumn stores the instance key without its escaping: the instance written as LineItems/item%2F7 is stored as item/7.

The instance table's primary key must be exactly the store's primary-key columns plus instanceKeyColumn, all character columns. With order-{customer_id}-{order_id}, the LineItems table needs the primary key customer_id, order_id, line_item_id. Every other required column must have a default or generated value. Dex checks the instance tables at startup and at every schema refresh, like the store's own table.

Writing an instance replaces its row, and deleting the instance deletes the row. An instance is one complete value, so its row always equals the latest value:

A JSON object writes each top-level field to the column with the same name, under columnNaming. A JSON number is written as an integer only where the column takes one. Nested objects and arrays go to JSON or JSONB columns.
Any other value is written to a column named value.
Every other non-key, non-generated column that the value does not set returns to its default, or NULL without one.
A field without a column is skipped. An instance table may store only part of each value, so Dex logs a warning once per process for each such field.
A field that maps to a primary-key column is skipped. Dex logs an error only when its value differs from the key.
A value that does not convert is logged as an error and skipped. The row still records the instance.

One batch writes the store row and all its instance rows in one transaction, with the latest value of each instance. Dex creates no foreign keys between the tables: a violated constraint would fail the whole batch and hold back the Flow's later writes until the retry budget runs out. Dex never deletes rows when a Flow closes.

A Flow that selects several stores writes each AttributeMap only to the stores that declare it. Other stores treat its instances as ordinary Attributes.

SQL writes and schema refresh

Each batch from one Flow is one upsert. With a rowKey, the upsert writes every key column and resolves conflicts on the whole primary key. Dex does not combine batches from different Flows and does not maintain a change log.

Each store refreshes independently. The one-minute default receives uniform 90–110% jitter every cycle. A successful metadata query atomically replaces the immutable in-memory schema snapshot. A failed query retains the previous snapshot and retries after the next interval.

Adding nullable or defaulted columns is discovered without restarting the server. A batch that writes a column missing from the current snapshot, in the store's table or an AttributeMap table, refreshes that store's schema before it writes. A column that no refresh has found missing refreshes at once; a column that a refresh already found missing waits for the periodic refresh. An AttributeMap table may store only part of each value, so each field without a column triggers at most one refresh per process. Batches that arrive meanwhile wait for that refresh. A migration that adds a column therefore needs no server restart, as long as it completes before the Workers that write the column start. Removing or changing columns may cause later mutations to be filtered or retried, depending on whether local validation or SQL execution detects the change.

SQL type contract

Attribute value	PostgreSQL column
string	Character, varchar, or text; declared length is enforced
int64	Signed integer or scale-zero numeric; range and precision are enforced
double	Floating or numeric
bool	Boolean
SDK datetime	Timestamp with time zone
JSON object	JSON or JSONB after validation
other object	bytea
null	Nullable columns only

SDK datetime values may use RFC3339 strings or JSON epoch seconds. Dex rejects oversized strings or binary values, out-of-range integers, excess decimal precision, non-finite floating-point values, and invalid JSON.

Missing columns, type mismatches, Flow ID and key column writes, Attribute names that share one column, and nulls for required columns are logged and filtered one field at a time. Valid fields in the same batch still write. If every field is filtered, the sync batch succeeds without executing SQL.

Behavior​

Configuration​

Fencing table​

Write timeout​

Retries and dropped entries​

Monitoring​

Startup contracts​

Row keys and column naming​

AttributeMap tables​

SQL writes and schema refresh​

SQL type contract​