Runtime state in configs

Morpheus is stateless between invocations except for one table. configs is a key/value store that holds the small amount of state a scheduled job needs to know where it left off.

Everything in it is one of three things:

Watermarks

"I have processed every record updated before this timestamp." Advanced after a successful sync, so the next run only reads what changed.

Cursors

"I stopped at this position." A row offset into a paginated API, or an index into an ordered list of sales reps.

Kill switches

"Do not run this." Set by hand to stop a job without a deploy.

There is no other runtime state. No DynamoDB table, no S3 marker object, no Parameter Store entry — a job’s progress is a row in the same transaction as the work it did, which is what makes the pipeline restartable.

The table

libs/morpheus-data-layer/morpheus_data_layer/models/configs.py, table configs.

Column Type Notes

id

UUID

Primary key, uuid.uuid4 default

key

String

Globally unique — not unique per subscription

subscriptionId

UUID

Tenant column; no ORM foreign key

value

String

Opaque; the reader parses it

createdAt

DateTime(timezone=True)

server_default=now()

updatedAt

DateTime(timezone=True)

server_default=now(), onupdate

Indexes: configs_key_idx, configs_subscription_id_idx.

key carries a global unique constraint, not a composite (subscriptionId, key) one. Two subscriptions cannot hold the same key. Production runs a single tenant, so this has never bitten — but any key that must be per-tenant has to embed the subscription id in the key string itself, which is exactly what the round-robin cursor does.

value is always a string. There is no JSON column and no type discriminator: the handler that writes a key is the only thing that knows how to read it.

Reading and writing

Through ConfigsRepository — see Repositories.

from morpheus_data_layer.repositories import ConfigsRepository

_configs_repo = ConfigsRepository()

with get_session() as session:
    raw = _configs_repo.get_value(session, WATERMARK_KEY)          (1)
    watermark = parse(raw) if raw else DEFAULT_WATERMARK           (2)

    rows = fetch_since(watermark)
    process(rows)

    _configs_repo.set_value(session, WATERMARK_KEY,                (3)
                            serialise(new_watermark), subscription_id)
1 Returns None when the key has never been written.
2 Every reader supplies a default. A missing key must never crash a handler — it is the normal state on a first run.
3 set_value() upserts on key, updating only value.

The commit is the caller’s. Because the watermark advance and the work share one transaction, a failure mid-batch rolls both back and the next run reprocesses the window. Every consumer of a watermark is therefore written to be idempotent — the engagement sync upserts on (subscriptionId, externalId), the call sync upserts on providerCallId.

Writing the watermark in a separate transaction from the work would let the two diverge: an advanced watermark plus a rolled-back batch silently loses records. Keep the set_value() call inside the same with get_session() block as the writes it describes.

Keys in use

Key Kind Written by Value format

enrichment.<marketplace>.disabled

Kill switch

Set by hand

"true" (case-insensitive) disables; anything else or absent means active

enrichment.<marketplace>.next_start_row

Cursor

enrichment_scraper/services/config_state.py

Integer as a decimal string

bigquery_engagement_sync_watermark

Watermark

morpheus_hubspot/events/scheduled/bigquery_engagement_sync.py

%Y-%m-%dT%H:%M:%S%z

five9_call_sync_watermark

Watermark

morpheus_hubspot/events/scheduled/five9_call_sync.py

ISO-8601 with offset

bq_contact_sync_watermark

Watermark

enrichment_scraper/events/bigquery/contact_sync.py

ISO-8601 from datetime.isoformat(), microsecond precision

dialer_follow_up_rep_rr:<subscriptionId>

Cursor

morpheus_data_layer/repositories/sales_rep_round_robin.py

Non-negative integer as a string

enrichment.<marketplace>.disabled

Read by ConfigStateService.is_disabled() and checked by the Amazon discovery handler before it calls SmartScout. Nothing in the codebase writes it — it exists to be set manually:

INSERT INTO configs (id, key, value, "subscriptionId")
VALUES (gen_random_uuid(), 'enrichment.amazon.disabled', 'true', '{subscription-id}')
ON CONFLICT (key) DO UPDATE SET value = 'true';

This is the fastest way to stop spending SmartScout quota. It takes effect on the next scheduled run with no deploy and no schedule change, and re-enabling is the same statement with 'false'. See Operations and runbooks.

enrichment.<marketplace>.next_start_row

The SmartScout pagination offset. ConfigStateService.get_next_start_row() reads it, save_next_start_row() advances it after a successful page. Discovery walks the seller catalogue linearly rather than re-querying from zero, so each run covers new ground.

Setting it back to "0" restarts the walk from the beginning — safe, because every downstream write is an upsert on sellerId, so a replay refreshes rows instead of duplicating them.

Both scraper keys are templated on marketplace, so enrichment.walmart.next_start_row is valid. Only amazon is used in production; the Walmart discovery handler is a stub.

The three watermarks

All three follow the same shape — read, default if absent, query "updated since", advance — but differ in their defaults and precision.

Key Default when absent Source it paginates

bigquery_engagement_sync_watermark

2026-05-15T00:00:00Z

hubspot4.engagements, filtered on lastupdated

five9_call_sync_watermark

2026-01-01T00:00:00+00:00

Five9 call export

bq_contact_sync_watermark

2026-01-01T00:00:00Z

hubspot4.contacts, filtered on updatedat

The defaults are backfill floors, chosen so a first run in an empty environment imports a useful history rather than everything ever recorded. They are not "the beginning of time" on purpose: hubspot4 holds years of engagements, and an unbounded first query would time out.

BigQuery is a periodic export, so a row’s lastupdated can be older than the moment it appears in the warehouse. A watermark advanced to exactly the last-seen timestamp can skip records that land late. Both BigQuery syncs deduplicate on the target key ((subscriptionId, externalId) for activities, contact identity for contacts), so an overlapping window is harmless — which is what makes it safe to rewind a watermark by a few hours when you suspect a gap.

Rewinding is the standard repair for a missed window:

UPDATE configs
SET value = '2026-08-01T00:00:00+0000'
WHERE key = 'bigquery_engagement_sync_watermark';

The next scheduled run reprocesses everything after that instant and upserts over what is already there.

dialer_follow_up_rep_rr:<subscriptionId>

The only per-tenant key, and the only one that embeds its subscription id in the key string — a workaround for the global unique constraint on key.

The value is an index into the ordered list of active sales reps returned by list_active_for_subscription() (ordered id ASC). reserve_sales_rep_round_robin_batch() takes a transaction-level advisory lock, reads the index, hands out count reps from that position, and writes the index back advanced by count.

The lock is what makes concurrent handlers safe: four re-engagement schedules run on the same rate(6 hours) cadence and would otherwise read the same cursor and assign the same rep to every batch. Details in Ownership and round-robin.

Deleting the key resets assignment to the first rep. Adding or deactivating a rep shifts the ordering under the cursor, which skews one cycle’s distribution and then self-corrects.

Inspecting state

SELECT key, value, "updatedAt"
FROM configs
WHERE "subscriptionId" = '{subscription-id}'
ORDER BY "updatedAt" DESC;

updatedAt is the useful column in an incident: a watermark whose updatedAt is hours stale means the schedule that owns it has not completed successfully, and that is usually a faster signal than reading logs.

Adding a key

  1. Define the key as a module-level constant next to the handler that owns it — never inline a string literal at the call site.

  2. Choose a namespaced name. Dots for scoping (enrichment.amazon.next_start_row), a colon for a tenant suffix.

  3. Give the reader a sensible default for the absent case. It must not raise.

  4. Serialise deterministically — a fixed strftime format or isoformat(), never str() on a datetime.

  5. Write it in the same transaction as the work it describes.

  6. Add a row to the table on this page.