DataCollector: Introduction & Architecture

📊 What is DataCollector?

DataCollector (DC) is EmpowerNow's enterprise-grade data synchronization engine that fetches data from configured systems via their APIs, stores entities per-system in PostgreSQL, and publishes diffs to EmpowerID, Kafka, or local files.

It uses a plugin architecture where:

• Strategies define collection logic (what to collect and when)
• Connectors handle protocols (SCIM, LDAP, REST, etc.)
• Schema Mappings transform API responses to standard entity structures

Each system declares one or more streams — named partitions that run inventories independently with their own scheduling and state. Multiple DC instances coordinate via per-stream database leases for high availability. When a downstream consumer falls behind, replay re-publishes stored entity data through the normal diff pipeline to bring the consumer back in sync.

🏗️ System Architecture

DataCollector integrates with the broader EmpowerNow ecosystem, coordinating with databases, secrets management, monitoring infrastructure, and downstream consumers.

Component Roles

Architecture Diagram

The following diagram illustrates how DataCollector instances coordinate with databases, secrets management, monitoring infrastructure, and downstream consumers. Note that work is partitioned by stream — different streams of the same system can run on different instances.

📋 Configuration Hierarchy

DC's configuration model separates concerns and enables reuse. Six layers, from least to most instance-specific:

1. Strategy — business logic implementation

Python plugin defining collection logic:

• Universal Strategy: builtin.strategies.universal.UniversalStrategy — YAML-driven collection with no Python code required. Handles standard API patterns purely via YAML configuration.
• Custom Python strategies: hardcode collection logic in Python for complex scenarios not expressible in YAML, providing optimized logic and patterns.

2. System Type — shared configuration template

Reusable YAML template defining shared defaults:

• Connector references and schema mapping names
• Pagination sizes and timeouts
• Schedule configuration (interval, full-sync threshold, delta lookback)
• Default stores and storage settings
• Default monitoring exclusions and circuit-breaker config

3. System — instance configuration

Instance YAML with:

• Credential pointer (vault URI like openbao+kv2://secret/datacollector/sapias-prod)
• Host / tenant / base_url (connection details)
• resource_system_id for EmpowerID sync
• Optional overrides of system type defaults
• Variables (for universal-strategy systems — values for the variables_schema declared in the system type)

4. Connector — protocol implementation

Python plugin implementing the wire protocol. Built-in connectors include:

• SCIM — SCIM 2.0 with pagination
• LDAP — LDAP search with cookie-based pagination
• REST — generic HTTP/REST
• AWS, Google, Echo, FlatFile — additional protocols

The multi-method pattern is common: a single connector exposes multiple operations (e.g., search / open_connection / paged_search / close_connection) via a method parameter in the params dict.

5. Schema Mapping — transformation rules

YAML transformation rules mapping connector response fields to entity fields. Three expression types:

• Direct — dot-path field references like result.id, pair.raw['email']
• Expression — Python wrapped in [[ ]] or marked with !expr tag
• String Template — Jinja {{ }} templates for string concatenation and formatting

Schema mappings are applied by DC (via the SchemaMapper service); strategies just reference them by name. They're reusable across multiple strategies and systems.

6. Credentials — authentication secrets

Auth secrets stored in vault (OpenBao/Vault):

• Referenced via credential_pointer URIs
• Contain only auth secrets — no connection details (host, port, base_url belong in system config)
• Connector-specific structures (LDAP bind DN/password, OAuth tokens, API keys)
• Rotate independently of configuration; DC resolves the pointer per connector call, so rotation takes effect on the next call

🎭 DC's Role & Boundaries

DataCollector orchestrates collection workflows but maintains clear boundaries between what it controls and what strategies and connectors control.

What DC validates vs. what's opaque

What DC enforces vs. what strategies decide

DC enforces:

• Step dependencies and execution order (by step name)
• Retries with optional per-instance and per-throttle-group backoff
• Throttle-group parallelism limits (see Throttle Groups)
• Resumable state persistence to database
• Circuit-breaker patterns for external calls
• Monitoring event publishing (Kafka, files)
• Per-stream lease coordination across DC instances

Strategies decide:

• Step logic and implementation
• Dispatch patterns (pagination, fan-out, nested fetching)
• Execution decisions based on previously-collected data
• Which connector to use (configured in YAML — strategies don't dynamically pick connectors at runtime)
• Their own config schemas via Pydantic models
• Stores, destinations, visibility rules, throttle-group definitions, replay profiles, commands

🌊 Streams

A stream is a named partition within a system that runs inventories independently. Each stream has its own scheduling, next_inventory_input, enable/disable status, and coordination lease.

Declaring streams

A strategy declares its streams via get_streams(). Every strategy must return at least one stream:

def get_streams(self, base_engine: BaseEngine) -> list[StreamDefinition]:
    return [
        StreamDefinition(name="users", entity_types=["account"]),
        StreamDefinition(name="groups", entity_types=["group", "membership"]),
    ]

The returned list must be the same for a given system + system-type configuration — it does not depend on next_inventory_input values. entity_types on each stream is for UI display grouping (many-to-many; not enforced by DC).

Per-stream methods

Most strategy methods receive stream_name to scope their behavior. Methods that operate system-wide do not.

Per-stream next_inventory_input (NII)

Each stream has its own NII stored in the database. NII is the primary channel for cross-inventory state — delta tokens, full-completed timestamps, failure counters, circuit-breaker state.

The snapshot model

When an inventory starts for a stream, the stream's NII is snapshotted onto the inventory record (inventory_input_snapshot). The running inventory reads this snapshot via ctx.inventory_input, so concurrent NII modifications (by other streams, by commands, or by the running inventory's own cross-stream writes) do not affect the running inventory's view. Editing NII while an inventory is in progress takes effect on the next inventory, not the current one.

Cross-stream NII writes

Steps can read and write any stream's NII during execution:

# Read another stream's current NII (live from DB, not snapshot)
groups_nii = await engine.read_stream_next_inventory_input("groups")

# CAS write to another stream's NII
engine.dispatch_stream_next_inventory_input(
    stream_name="groups",
    old_value=groups_nii,
    new_value={**groups_nii, "trigger_full": True},
)

Cross-stream writes are buffered on the step and applied at commit via compare-and-swap. If the CAS fails (another process modified the value), DC calls resolve_next_inventory_input_conflict() to reconcile — that method must be a pure data merge with no I/O and a 10-second timeout.

Per-stream control

Each stream can be independently:

• Enabled / disabled — disabled streams don't schedule inventories
• Triggered — trigger profiles are per-stream
• Stopped — stop a running inventory for a specific stream

System-level operations span streams: cleanup-entities (deletes entities and resets all streams' NIIs, requires holding all stream leases) and system-wide replay.

For full reference: strategies-development/045-streams.md.

🔄 Inventories & Trigger Modes

An inventory is one execution of data collection for a stream. Inventories are:

• Resumable — state persisted in database, can continue after DC restart
• Exclusive per stream — only one inventory per stream at a time, but a system with multiple streams runs them concurrently (possibly on different DC instances)
• Stateful — receive inventory_input snapshotted from the stream's NII at start

Five trigger modes

Effects matrix

Cross-inventory state via NII

Inventories pass data forward via next_inventory_input for state that crosses inventory boundaries:

• Delta sync timestamps — track last successful sync time for incremental fetches
• Full-completed timestamps — decide when to next run a full sync vs. delta
• Failure counters — implement adaptive scheduling
• Circuit-breaker state — remember system health across runs

Within-inventory pagination is not NII — pagination tokens belong in step inputs (dispatch_step parameters) or in the large-list cache. NII is for state across inventory boundaries. See Range Fan-Out & Cache below.

For full per-mode lifecycle, persistence rules, and failure behavior: system-admins/05-execution-modes.md.

🔁 Replay

DC normally assumes every published diff was processed by the consumer. When that assumption breaks — consumer downtime, inbox failure, processing error — the consumer's state drifts from DC's. Replay fixes this by re-publishing stored entity data through the normal diff pipeline. No connector calls, no strategy execution: DC reads attribute_transformed_data from its database and pushes it through the same diff pipeline that normal inventories use.

Two modes

• REPLAY — publishes diffs to destinations (File, Kafka, EmpowerID). The mode that actually fixes consumer state.
• READONLY_REPLAY — formats diffs and emits monitoring events but skips destination publishing. Preview what a replay would publish before committing.

Replay profiles

Strategies declare replay profiles via get_replay_profiles(). Each profile targets exactly one store (see Stores, Destinations & Output Modes below) and lists the destination routings to use. To replay multiple stores, trigger separate replays. The strategy controls which stores are replayable.

The since parameter

Optional timestamp. Controls which entities are replayed:

• Omitted — all entities in the store are replayed (full re-publish)
• Provided — only entities whose store data changed at or after that timestamp; recommended value is the start timestamp of the inventory whose diffs were not processed

EmpowerID compatibility (CREATED / UPDATED / DELETED contract)

The agreed contract:

• CREATED — processed as upsert (works whether the entity exists at the consumer or not)
• UPDATED — applied to an existing entity; skipped if the entity doesn't exist
• DELETED — idempotent; applied if the entity exists, no error if it doesn't

Replay uses created_or_resurrected_on to decide diff type. With since, visible entities established at the consumer before the replay boundary get UPDATED; entities the consumer may not have get CREATED; gone-from-store entities get DELETED. Without since (full replay), all visible entities get CREATED — safe at the consumer because CREATED is processed as upsert.

FK-consistent ordering

When visibility rules include ForeignKeyCheck (see Visibility Rules below), replay produces diffs in FK-consistent order so consumers never see a reference to a nonexistent entity:

• Alive pass — parents before children (CREATED / UPDATED diffs)
• Dead pass — children before parents (DELETED diffs)

For full reference (visible vs dead semantics, attribute-level tombstones in PER_ATTRIBUTE mode, coordination lease behavior): system-admins/04-replay.md.

⚡ Commands

Commands are user-invocable operations that modify a stream's next_inventory_input via compare-and-swap, without starting an inventory. Useful for administrative actions like "Force Full Sync", "Reset Delta Token", "Clear Failure Counter".

Declaring commands

def get_commands(self, next_inventory_inputs, base_engine) -> list[CommandDefinition]:
    return [
        CommandDefinition(
            name="force_full_sync",
            description="Reset delta tokens and force a full sync next run",
            stream_name="users",  # appears as a button on the users-stream row
        ),
        CommandDefinition(
            name="reset_all",
            description="Clear all state across streams",
            stream_name=None,  # system-level button
        ),
    ]

Executing commands

def execute_command(self, command_name, next_inventory_inputs, command_engine) -> CommandResult:
    if command_name == "force_full_sync":
        nii = next_inventory_inputs["users"]
        command_engine.dispatch_stream_next_inventory_input(
            "users", old_value=nii, new_value={**nii, "force_full": True},
        )
        return CommandResult(success=True, message="Full sync will run next")

Commands:

• Run any time, including during active inventories (CAS handles conflicts)
• Do not start inventories, do not acquire stream leases
• Use CommandEngine (limited to NII read/write and template evaluation — no call_connector, no dispatch methods)
• Appear in the UI as buttons on stream rows (if stream-scoped) or system rows (if stream_name=None)
• Result is shown to the user via a CommandResult with a success flag and a message

For full reference: strategies-development/045-streams.md.

🗄️ Stores, Destinations & Output Modes

DC's model for change tracking and publishing has two layers: stores (DC's record of what each consumer knows) and destinations (transports that deliver diffs). They are not the same thing.

Stores: named state slots

A store is a per-system, named slot that tracks per-entity state in the database: an entity-level hash, per-attribute hashes, per-attribute timestamps, transformed data, and a visibility flag. Each store operates independently — the same entity can be visible in store "eid" and invisible in store "kafka-feed", with different hashes and different transformed data.

Each entity's state in a store is one of three:

• Visible — entity alive and published to consumers
• Invisible — entity hidden from consumers (appears deleted to them) but data retained for potential resurrection
• Deleted — entity gone from this store, data cleared

Strategies declare stores via get_destination_config()

def get_destination_config(self, base_engine) -> Dict[str, StoreConfig]:
    return {
        "eid": StoreConfig(
            entity_types=["account", "group"],
            empowerid=[EmpowerIDStoreRouting(
                doc_type_ids={"account": 42, "group": 43},
                resource_system_id=self.resource_system_id,
                entity_message_formatter=empowerid1,
            )],
            kafka=[],
            file=[],
        ),
        "kafka-feed": StoreConfig(
            entity_types=["account"],
            empowerid=[],
            kafka=[KafkaStoreRouting(
                output_mode=OutputMode.PER_ATTRIBUTE,
                entity_types=["account"],
                topic=None,  # None → use DC default topic
                attribute_message_formatter=empowernow1_per_attribute,
            )],
            file=[],
        ),
    }

Destinations: transport types

Three destination types:

• EmpowerID — API sync, deduplicated by entity (last diff wins), batched by doc_type_id; always PER_ENTITY
• Kafka — streaming, no deduplication, message keyed by system_name
• File — local YAML files (debugging-friendly; also used as a fallback when DC config diverts other destinations)

A store routes to zero or more destinations. The same store's change-detection output can publish to multiple destinations with different shapes. A store with zero destinations tracks changes but publishes nowhere — useful for reserving a tracking slot for a future consumer.

Output modes per destination

Each Kafka or File route specifies an output_mode:

• PER_ENTITY — one diff per entity. CREATED/UPDATED include full transformed_data; DELETED carries entity_keys for the consumer to identify the gone entity.
• PER_ATTRIBUTE — one diff per changed attribute. When attributes are removed (or the entity is deleted), tombstone diffs are produced. Consumers can subscribe to specific attributes.

EmpowerID is always PER_ENTITY. Switching output modes is safe at any time — the store's tracking state (hashes) doesn't change, only how diffs are formatted.

Message formatters

Strategies own the message shape. Each route can specify entity_message_formatter (PER_ENTITY) or attribute_message_formatter (PER_ATTRIBUTE). The SDK provides reusable formatters:

• empowerid1 — for EmpowerID destinations (CREATED/UPDATED/DELETED with an Action field)
• empowernow1_per_entity — for Kafka/file PER_ENTITY (entity envelope with payload)
• empowernow1_per_attribute — for Kafka attribute-flow consumers

Formatters receive a diff (EntityDiff or AttributeDiff protocol) plus a MessageContext (system_name, inventory_id, inventory_run_id) and return a Dict[str, Any]. Strategies can override formatters per route or define their own.

DC-level overrides

DC config (in config.yaml) can modify routing globally per destination type:

• disable — destination removed from store routings
• mirror_to_file — destination stays active, File destination added alongside
• divert_to_file — destination removed, File destination added instead
• mirror_all_stores_to_file — File destination added to every store

These are operational toggles for testing and incident response, not strategy concerns.

For full reference: strategies-development/050-changes-processing-pipeline.md.

🏷️ Entity Attributes

Each entity has three optional attributes stored on the entity row: attribute1, attribute2, attribute3. They are nullable strings, non-unique, and indexed.

Use cases

• ETags — cheap freshness check before fetching full entity data
• Classification — group entities for batch processing
• Soft-delete flags — combined with AttributeCheck visibility rules (see below)
• Strategy-specific metadata — anything you'd want to look up later

Setting attributes

Attributes are set via dispatch_entity_confirmed_exact:

engine.dispatch_entity_confirmed_exact(
    entity_type="account",
    transformed_data=item.transformed,
    key1=user_id,
    attribute1=item.transformed['eid'].get('etag'),
    attribute2="active" if item.transformed['eid']['enabled'] else "soft_deleted",
)

Looking up by attribute

entities = engine.lookup_entities_by_attribute(
    entity_type="account",
    attribute1=etag_value,  # only non-None values are used as filters
)

For full reference: strategies-development/085-entity-lookup-and-dispatched-changes.md.

👁️ Visibility Rules

Not all source systems provide FK-consistent data. An API may return a group membership before the group itself, or delete a parent entity while children still reference it. Visibility rules let strategies declare FK relationships; DC enforces referential integrity before publishing diffs to consumers.

Rules are returned by get_visibility_rules() as a flat list. Each rule targets a specific (store_name, entity_type) pair. An entity is invisible in a store if any of its rules evaluates to invisible.

Two rule types

AttributeCheck

Entity invisible when an entity-level attribute equals a specific value:

AttributeCheck(
    store_name="eid",
    entity_type="account",
    attribute_name="attribute1",
    value="soft_deleted",  # entity invisible when attribute1 == "soft_deleted"
)

ForeignKeyCheck

Entity invisible when the referenced entity is not alive in the same store:

ForeignKeyCheck(
    store_name="eid",
    entity_type="membership",
    relation_key_number=1,         # uses relation1_key1 → key1 of referenced
    referenced_entity_type="group",
)

Backward cascade is automatic

When entity U changes state, DC finds entities referencing U via relation keys, re-evaluates their visibility, and produces appropriate diffs. The strategy declares the rules; DC handles the cascade.

• V→I (was visible, now invisible) → DELETED diff with decision "BecameInvisible"
• I→V (was invisible, now visible) → CREATED diff with decision "BecameVisible"
• Cascade through dependent entities → "CascadeBecameInvisible" / "CascadeBecameVisible"

Constraints

• No same-type references (entity_type must differ from referenced_entity_type)
• FK dependency graph must be acyclic — DC validates at inventory start

Consumer perspective

Invisible entities appear deleted to consumers. Data is retained on the DC side, so when the entity becomes visible again, the CREATED diff carries the latest data, not stale data from the moment it became invisible. Replay treats only visible entities as present; invisible entities are replayed as DELETED tombstones.

For full reference: strategies-development/090-visibility-rules.md.

🚦 Throttle Groups

Fan-out strategies dispatch many step instances that all hit the same external API. Without coordination, all instances retry independently after a failure wave — collectively overwhelming the server. Throttle groups coordinate parallelism and backoff across step instances sharing a resource.

Declaring group membership

Steps declare group membership via throttle_groups in StepExecutionConfig:

async def get_step_execution_config(self, step_name, stream_name, context, engine):
    if step_name == "fetch_user":
        return StepExecutionConfig(
            max_retries=5,
            throttle_groups=["ldap_server", "all"],
        )
    return None

A step can belong to multiple groups; all group constraints must be satisfied (AND logic).

Defining group config

async def get_throttle_group_config(self, group_name, stream_name, base_engine):
    if group_name == "ldap_server":
        return ThrottleGroupConfig(
            max_parallelism=3,
            retry_backoff=RetryBackoff(
                initial_delay_seconds=2.0, factor=2.0,
                max_delay_seconds=90.0, jitter=0.3,
            ),
        )
    if group_name == "all":
        return ThrottleGroupConfig(max_parallelism=10)
    raise NotImplementedError(f"Unknown throttle group: {group_name}")

How it works

DC divides each group into max_parallelism virtual slots. Slot state is derived from step_runs in the database (not in-memory). Each slot independently tracks consecutive failures (slot_level); on RETRYABLE_FAILURE, the slot enters backoff with a delay computed as initial × factor^slot_level, capped at max_delay, with deterministic per-slot jitter so all slots don't expire simultaneously.

Recommended default pattern

For most fan-out strategies hitting one shared upstream resource:

• Put all relevant steps in a shared throttle group
• Set ThrottleGroupConfig.max_parallelism to limit concurrent callers
• Set ThrottleGroupConfig.retry_backoff to protect the resource (slows new first attempts too — group backoff applies to every member, not just retries)
• Also set StepExecutionConfig.retry_backoff so each individual failing step is paced predictably

DC enforces the longer of the two backoffs.

For full reference (multi-group AND, virtual slot model, gradual recovery, jitter formula): strategies-development/100-throttle-groups.md.

📤 The Dispatch Interface

Strategies signal entity state changes to DC via the dispatch interface — eight typed methods on InventoryEngine. Each method says what happened, not what to do about it. The strategy provides keys and data; DC handles diff types, visibility, hashing, formatting, and publishing.

The eight methods

Universal-strategy YAML exposes 4 of 8

The Universal Strategy's produces: blocks expose only four:

• confirm → dispatch_entity_confirmed_exact
• delete → dispatch_entity_deleted
• purge_unseen → dispatch_purge_unseen
• purge_by_relation → dispatch_purge_unseen_by_relation — note that the YAML keyword drops the unseen_, but the underlying engine call is the unseen-since-cutoff variant, not an unconditional purge

Strategies needing the other four (dispatch_purge and dispatch_purge_by_relation — the two unconditional purges, plus dispatch_existing_entity_seen and dispatch_existing_entity_confirmed_partial) must be written as custom Python strategies.

min_absent_inventories deferral

dispatch_purge_unseen and dispatch_purge_unseen_by_relation accept an optional min_absent_inventories threshold (default 1). Entities below threshold get their absent_inventory_count incremented and are deferred rather than purged. Useful for tolerating non-snapshot-consistent pagination — an entity may be missed in a single enumeration pass but is not actually deleted.

engine.dispatch_purge_unseen(
    entity_type="account",
    min_absent_inventories=3,  # purge only after absent for 3 consecutive inventories
)

since cutoff for purge

Purge methods also accept an optional since datetime — entities with last_confirmed_on before this time are purge candidates. Default is the inventory's start time.

For full reference: strategies-development/005-strategy-dc-boundary.md, strategies-development/020 sdk, engine, inventory execution overview.md.

🎯 Step Run Outcomes

Every step execution produces an outcome. DC has six outcome values, each with specific persistence rules. Inventories don't have a formal success/failure status — they always complete. Outcomes affect what each step persists, not the inventory as a whole.

Six outcomes

Persistence rules — the central rule

Only SUCCESS and EXPECTED_FAILURE persist a step's changes and dispatched steps. All other outcomes discard them, with one exception (see dispatch_step_on_fatal_failure below).

Dependencies don't auto-fail — a step with a FATAL_FAILURE dependency still runs (the dependent step receives an unblocked terminal state, not a propagated failure). If you want cascading failures, check the dependency's outcome via get_step_statistics().

The FATAL_FAILURE exception: dispatch_step_on_fatal_failure

Sometimes you want a cleanup or alert step to run only if the current step fails fatally. Use engine.dispatch_step_on_fatal_failure(name, parameters): the dispatched step is persisted only when the current step ends with FATAL_FAILURE (and discarded on all other outcomes including SUCCESS). Works on every FATAL_FAILURE path: explicit finish_step(FATAL_FAILURE), unhandled exceptions, commit-phase failures, and max_retries exhaustion.

Setting outcomes

SUCCESS is automatic if the step finishes without an explicit finish_step() call. All other outcomes must be explicit:

if result.status == ConnectorCallStatus.NOT_FOUND_ERROR:
    engine.finish_step(Outcome.EXPECTED_FAILURE, "Resource not found, skipping")
    return

if result.status == ConnectorCallStatus.THROTTLED_ERROR:
    engine.finish_step(Outcome.RETRYABLE_FAILURE, "Rate limited")
    return

if result.status != ConnectorCallStatus.SUCCESS:
    engine.finish_step(Outcome.FATAL_FAILURE, f"Failed: {result.error_data}")

For full reference (unique-key behavior, dependency model, stuck-inventory detection): strategies-development/030-inventory-flow-and-state.md.

🚀 Eager Dispatch & DISCARDED

By default, dispatched steps don't start until the dispatching step commits. Eager dispatch (immediate=True) starts the dispatched step before the parent commits, overlapping the child's connector call with the parent's remaining work. Useful for latency-sensitive fan-out where waiting for a slow commit is wasteful.

engine.dispatch_step("fetch_user_details", parameters={"id": user_id}, immediate=True)

If the parent fails

If the parent ends with FATAL_FAILURE or RETRYABLE_FAILURE, eagerly-dispatched children get the DISCARDED outcome — monitoring-only, never persisted to DB. The connector call already happened, but entity processing and DB commit don't run.

Restrictions (footguns)

• The child must not declare a step-name dependency on the parent. If it does, the dependency check blocks the child until the parent commits, silently degrading immediate=True to normal dispatch timing.
• The child must not depend on the parent's entity changes being in the DB (parent hasn't committed).
• The child must not depend on NII written by the parent (cross-stream NII writes happen during commit).
• The child must not rely on step statistics reflecting the parent's outcome.
• Connector call side effects cannot be undone on DISCARDED — the strategy developer's responsibility.
• immediate=True cannot be combined with unique_key — raises ValueError. Unique-key merge logic requires durable Step rows that don't exist during eager dispatch's transient phase.

For full reference: strategies-development/080-strategy-ground-rules.md.

🔍 Lookup Methods

Strategies can read entity state during step execution via four lookup methods on InventoryEngine.

Single entity by key

entity = await engine.lookup_entity(
    entity_type="account",
    source_key_number=1,
    source_key_value=user_id,
)
if entity is not None:
    # EntityInfo: key1-3, relation1-3_key1, attribute1-3, is_deleted
    if entity.attribute1 == "etag_v2":
        # cheap freshness check, skip full fetch
        engine.dispatch_existing_entity_seen(entity_type="account", key1=user_id)
        return

Multiple entities by attribute

entities = engine.lookup_entities_by_attribute(
    entity_type="account",
    attribute1="etag_value",
)

Multiple entities by relation

members = engine.lookup_entities_by_relation(
    entity_type="membership",
    relation1_key1=group_id,
)

Bulk key translation

id_map = await engine.lookup_entity_key(
    entity_type="account",
    source_key_number=1,  # search by key1
    target_key_number=2,  # return key2 values
    source_values=["a@x.com", "b@y.com"],
)

For full reference: strategies-development/085-entity-lookup-and-dispatched-changes.md.

🪟 Range Fan-Out & Cache

Each step must have a small, bounded memory footprint regardless of total system size. Strategies use two patterns to break large work into small steps: range fan-out via get_dispatched_changes, and the large-list cache for unpaginated large responses.

Range fan-out via get_dispatched_changes

Step A dispatches N sub-steps with {offset, limit} parameters. Each sub-step reads its slice of A's dispatches:

# Step A: confirms N entities, then dispatches sub-steps
for offset in range(0, total, 100):
    engine.dispatch_step(
        "process_batch",
        parameters={"offset": offset, "limit": 100},
    )

# process_batch step: reads its slice
@step
async def process_batch(self, ctx, engine):
    changes = engine.get_dispatched_changes(
        step_name="step_a",
        offset=ctx.step_input["offset"],
        limit=ctx.step_input["limit"],
    )
    for change in changes:
        # process each one

Results from get_dispatched_changes have stable, deterministic ordering, so offset/limit pagination is safe across multiple calls within the same step execution. Step B must declare a dependency on Step A so A's changes are committed before B reads them.

Large-list cache

If the source API returns all entities in a single unpaginated response, save the response to cache immediately and dispatch processing steps that read from cache in ranges:

cache_id = await engine.save_to_cache(items)
for offset in range(0, len(items), 1000):
    engine.dispatch_step(
        "process_chunk",
        parameters={"cache_id": cache_id, "offset": offset, "count": 1000},
    )

# process_chunk step
chunk = await engine.load_from_cache(
    cache_id=ctx.step_input["cache_id"],
    start_index=ctx.step_input["offset"],
    count=ctx.step_input["count"],
)

For full reference: strategies-development/085-entity-lookup-and-dispatched-changes.md, strategies-development/080-strategy-ground-rules.md.

📡 Connector Call Results

engine.call_connector() returns a ConnectorCallResult with a typed status field. There are eleven status values across three categories.

The full enum

Categories

• Remote — connector interpreted the response from the external system
• Infrastructure — DC engine produced this before or instead of calling the connector
• Internal — bug in our code that needs fixing

Result structure

On SUCCESS:

• transformed_data: Dict[str, List[...]] — per-store transformed lists
• correlations.raw_connector_response: original raw response
• correlations.transformed_items: per-item correlation between raw and per-store transformed
• additional_output: metadata extracted by the optional additional_output callable (e.g., pagination tokens)
• duration_ms: elapsed time

On any error status: error_data contains connector-specific error details, correlations is None.

Typical handling

Most strategies don't distinguish between failure types:

result = await engine.call_connector(...)
if result.status != ConnectorCallStatus.SUCCESS:
    engine.finish_step(Outcome.RETRYABLE_FAILURE, f"Connector failed: {result.status.value}")
    return

When a strategy needs to react differently:

if result.status == ConnectorCallStatus.AUTHENTICATION_ERROR:
    engine.finish_step(Outcome.FATAL_FAILURE, "Credentials rejected by remote system")
    return
if result.status == ConnectorCallStatus.NOT_FOUND_ERROR:
    engine.finish_step(Outcome.EXPECTED_FAILURE, "Resource not found, skipping")
    return

For full reference: strategies-development/020 sdk, engine, inventory execution overview.md.

🪜 Engine Tiers

The SDK provides five engine tiers at different lifecycle phases. Each tier exposes only what's needed for that phase — the type system enforces what's available when. You can't call call_connector() during trigger evaluation, for example.

For full reference: strategies-development/020 sdk, engine, inventory execution overview.md.

❤️ Health Checks

Strategies declare what to check; DC handles how. Health-check results are surfaced as 24-hour mini-graphs on the Inventories page (per-system) and the Credentials list (per-credential).

Strategy contract

def get_health_check_details(self) -> List[HealthCheckDetails]:
    return [HealthCheckDetails(
        connector_name="builtin.connectors.scim.ScimConnector",
        credential_pointer=self.credential_pointer,
        params={"base_url": self.base_url, "endpoint": "/ServiceProviderConfig"},
        auto_timeout_seconds=15.0,
        manual_timeout_seconds=60.0,
    )]

A system can have multiple health checks (one per connector). Testing a credential pointer runs only checks that use that credential.

Two timeout configurations

• auto_timeout_seconds — capped timeout for scheduled automatic checks; prevents scheduler overrun
• manual_timeout_seconds — uncapped timeout for on-demand testing via the UI; allows thorough validation

Default automatic schedule: every 15 minutes.

💾 Storage Configuration

Storage controls what gets persisted to PostgreSQL per step. Strategies return a StorageConfig from get_step_storage_config(step_name, stream_name, base_engine). Different steps in the same inventory can have different storage configs.

Three controls

• disable_changes — entity types to skip storing StepRunChange records (saves space; no history of dispatches for those types)
• enable_entities_transformed_data — entity types to store full transformed data on the entity row (required for dispatch_existing_entity_confirmed_partial and for replay readiness)
• enable_changes_transformed_data — entity types to store full transformed data on each change record (for audit and get_dispatched_changes(include_transformed_data=True); significantly increases DB size)

Hashes are always stored regardless of transformed_data policy — change detection works without full data.

🛡️ Monitoring & Circuit Breaker

Monitoring

DC publishes monitoring events to Kafka (the datacollector-monitoring topic) and optionally to local YAML files. Configuration controls which events and fields are excluded:

• Disable entire event types (e.g., template_evaluation events to reduce volume)
• Disable specific fields within events (hide sensitive data)
• Per-destination rules (different rules for Kafka vs. files)
• Contextual filters (decision_unchanged for change events, template_use_case_regex for template evaluations)

Circuit Breaker

Protects external system calls by opening after repeated failures, causing subsequent calls to fail fast without waiting for timeouts, then testing recovery after a timeout period.

Four ConnectorCallStatus values trigger the circuit (see Connector Call Results for the full enum):

• THROTTLED_ERROR (429 rate limit)
• CONNECTIVITY_ERROR (network failure, DNS, connection refused)
• TIMEOUT_ERROR (request timeout)
• AUTHENTICATION_ERROR (credentials rejected — likely systemic)

Other errors don't trigger:

• BAD_REQUEST / NOT_FOUND_ERROR — request-specific, not system-wide
• SERVER_ERROR — may be request-specific; default to not triggering to avoid false opens
• FATAL_ERROR — connector or DC bug, not external system issue

Hierarchical configuration

Circuit breaker config merges field-by-field across three levels: global (DC config) → system type → system. Each field independently overrides — a system can override threshold without specifying timeout, inheriting the latter from the system type.

⚖️ Horizontal Scaling

DC achieves high availability and load distribution through horizontal scaling — multiple DC instances coordinate via database leases to ensure work is distributed without conflicts.

Per-stream lease coordination

Each stream has its own lease (not each system). A multi-stream system can have its streams running on different instances simultaneously.

• Acquisition — instance attempts an atomic lease insert before starting work; the database UNIQUE (system_name, stream_name) handles the race
• Heartbeat — owning instance updates heartbeat_at on a fixed interval with version-based optimistic locking (default heartbeat_interval_seconds: 5)
• Local expiration check — owning instance treats the lease as expired if its last heartbeat started longer ago than heartbeat_failure_timeout_seconds (default 30s; Python time, clock-drift immune)
• DB expiration — other instances can steal the lease once its heartbeat_at is older than lease_timeout_seconds (default 60s)

The three values are configurable in the coordination block of DC's config.yaml; the relationship lease_timeout > heartbeat_failure_timeout > heartbeat_interval × 2 is enforced at construction. Production deployments often raise the two timeouts well above their defaults to tolerate slow restarts and long-running operations.

Coordination lease (status coordinator)

A separate STATUS_COORDINATOR coordination lease (independent of stream leases) elects one instance to poll system config files and update statuses. Prevents duplicate work and races in system discovery.

Replay's CompoundLeaseInfo

Replays operate at the system level (re-publish data spanning entity types from multiple streams), so they acquire all stream leases for the system at once via CompoundLeaseInfo. Stream-level operations (cleanup-ongoing-inventory, edit NII, stop-inventory) acquire only the relevant stream's lease.

Failure scenarios

• Concurrent triggers — multiple instances try to acquire the same stream lease; database constraint ensures only one succeeds; losers get None (expected contention, not an error)
• Network partition — instance A loses connection, heartbeats fail; after 30s A detects local expiration and terminates its inventory; after 60s instance B can steal the lease and resume
• Instance crash — lease expires after 60s; another instance takes over and resumes the inventory's persisted state from where it stopped

🔐 Authentication & Authorization

DC runs in one of three authorization modes, selected at startup by config.auth.mode. Two endpoints bypass both layers: GET /health and GET /monitoring/schema.

Three modes

auth.mode is mandatory in config.yaml. There is no default — a missing or misspelled value fails startup.

Scope mode

The IdP issues a token carrying scopes. DC requires:

• datacollector:read for read endpoints (system list, entity browse, config read, health-check)
• datacollector:write for mutations (enable/disable, trigger, edit configs, command execution)
• datacollector:debug for debug endpoints (POST /debug/strategy, POST /debug/strategy/stream)

Missing scope returns 403 insufficient_scope.

PDP mode

DC sends an AuthZEN request per endpoint. The PDP returns permit or deny. The dc-ui PermissionGrants are roles like:

• app_user — view permissions only
• app_admin — view + manage non-strategy resources
• platform_admin — full access including strategy management and debug

Trusted services + X-Actor-ARN

When a known service (e.g., CRUDService MCP) calls DC on behalf of a real user, the service forwards an M2M token plus an X-Actor-ARN header carrying the real user's canonical ARN. DC's auth middleware sees the service in auth.trusted_services and swaps the token subject for the actor ARN. The PDP evaluation runs against the real user, so audit events show the user, not the service.

Fail-closed: if the trusted-service token arrives without a canonical X-Actor-ARN, DC returns 401 actor_arn_required. This prevents privilege escalation through borrowed service tokens.

For full reference (troubleshooting 401/403, env var configuration, the AuthZEN request shape): system-admins/06-authentication-and-authorization.md.

📝 Five Development Considerations

When designing a strategy, consider these five aspects (from strategies-development/010-inventory-strategies-development.md):

1. Data consistency during inventory execution — does your strategy handle source data changes gracefully during inventory execution? Ensure no false deletions occur when data is modified mid-collection.
2. Resumability — can the inventory resume properly if DataCollector is restarted? Essential for long-running strategies where DC may be interrupted during long inventories.
3. Key change detection — are key changes properly detected and handled? Your strategy should identify when entity keys are modified.
4. Key reuse scenarios — if a key changes and another entity adopts the previously-used key, is this scenario correctly detected?
5. Pagination and source data integrity — does the source API guarantee snapshot isolation? Can entities be missed or duplicated if data changes mid-pagination? Use min_absent_inventories deferral on purges to tolerate non-snapshot pagination.

📐 Strategy Ground Rules

Headlines from strategies-development/080-strategy-ground-rules.md:

• Data sync guarantee — DC's purpose is to guarantee entity state matches the source. When in doubt, choose correctness over speed.
• Single file, no code sharing — a strategy is a single Python file. No imports from other strategies.
• No static or class-level state — steps may run on different pods. All inter-step communication via DB-persisted channels (step inputs, NII, cache).
• Step input size limit — 1KB serialized. Step inputs are coordinates into the work, not the work itself.
• Step memory footprint — each step bounded regardless of total system size. One step = one connector call (or a small number of related calls). Pagination = next page as a new step.
• Fail fast — never silently skip data, swallow errors, or continue with partial results. Prefer a failed inventory over a silently incomplete one.
• Call ordering within a step — all engine methods must be called before finish_step. After finish_step, any engine call raises.
• Lookup read isolation — structure step dependencies so that data the strategy reads is fully committed before the reading step begins.
• NII resolver must be fast — resolve_next_inventory_input_conflict() has a 10-second timeout; pure data merge, no I/O.
• Caching determinism — config-method results are cached by DC; cached methods must return the same result for the same configuration.