# Delta-Sync Sentinel (`automationnation/delta-sync-sentinel`) Actor RAG Pipeline, Agentic Infrastructure, Idempotent Ingestion — transactional-outbox delta sync engine with deterministic chunk IDs and lease/fencing tokens. Safe to re-run; manifest-commit is the source of truth. - **URL**: https://apify.com/automationnation/delta-sync-sentinel.md - **Developed by:** [Nathan Carter](https://apify.com/automationnation) (community) - **Categories:** Developer tools - **Stats:** 1 total users, 0 monthly users, 0.0% runs succeeded, 0 bookmarks - **User rating**: No ratings yet ## Pricing from $0.15 / actor start This Actor is paid per event and usage. You are charged both the fixed price for specific events and for Apify platform usage. Since this Actor supports Apify Store discounts, the price gets lower the higher subscription plan you have. Learn more: https://docs.apify.com/platform/actors/running/actors-in-store#pay-per-event ## What's an Apify Actor? Actors are a software tools running on the Apify platform, for all kinds of web data extraction and automation use cases. In Batch mode, an Actor accepts a well-defined JSON input, performs an action which can take anything from a few seconds to a few hours, and optionally produces a well-defined JSON output, datasets with results, or files in key-value store. In Standby mode, an Actor provides a web server which can be used as a website, API, or an MCP server. Actors are written with capital "A". ## How to integrate an Actor? If asked about integration, you help developers integrate Actors into their projects. You adapt to their stack and deliver integrations that are safe, well-documented, and production-ready. The best way to integrate Actors is as follows. In JavaScript/TypeScript projects, use official [JavaScript/TypeScript client](https://docs.apify.com/api/client/js.md): ```bash npm install apify-client ``` In Python projects, use official [Python client library](https://docs.apify.com/api/client/python.md): ```bash pip install apify-client ``` In shell scripts, use [Apify CLI](https://docs.apify.com/cli/docs.md): ````bash # MacOS / Linux curl -fsSL https://apify.com/install-cli.sh | bash # Windows irm https://apify.com/install-cli.ps1 | iex ```bash In AI frameworks, you might use the [Apify MCP server](https://docs.apify.com/platform/integrations/mcp.md). If your project is in a different language, use the [REST API](https://docs.apify.com/api/v2.md). For usage examples, see the [API](#api) section below. For more details, see Apify documentation as [Markdown index](https://docs.apify.com/llms.txt) and [Markdown full-text](https://docs.apify.com/llms-full.txt). # README ## Delta-Sync Sentinel **Stateful Data Observability & Ingestion Infrastructure.** Delta-Sync Sentinel is not a scraper — it's a transactional-outbox sync engine. It fetches sources, splits them into content-addressed chunks, and commits them as immutable **manifests** under a monotonic **cursor**. Every design decision below serves one property: **re-execution is harmless, and manifest-commit is the single source of truth.** That's what makes it safe to sit underneath an autonomous agent or a production ingestion pipeline, rather than just a one-off script. ### Consumer contract (read this first) > **Downstream consumers MUST upsert on `chunk_id`. Do not append.** - **Progress is a cursor, not a clock.** Track your position with the integer `cursor` (a monotonic counter), never with `committedAt`. The timestamp is informational only. - **Trust the commit boundary.** Only process chunks belonging to a manifest whose `cursor <= last_cursor` (see the health endpoint). Dataset rows with a `cursor` above the committed head are orphans from an in-flight or dead run and must be ignored. - **Chunks are content-addressed.** `chunk_id = sha256(source_url + section_path + content_hash)`. The same input always produces the same id, so upserting is idempotent and replays are free. ### Agentic Integration Contract This actor is designed to be driven by an autonomous agent, not just a human clicking "Start." An agent needs three things to do that safely: how to trigger a sync, how to recover from a gap, and how to tell if it's safe to trust what it's reading. The system prompt below covers all three — copy it directly into an agent's instructions. ```` You control the "Delta-Sync Sentinel" Apify actor. Follow this contract exactly: 1. To ingest new/changed sources, call the actor with: { "op": "sync", "sources": \[...urls], "maxCostUsd": "" } 2. Before trusting any dataset row, check the actor's health endpoint (GET /health in Standby mode, or read the run's OUTPUT key-value entry after an op:"health" call). Only trust rows where row.cursor <= last\_cursor. If status is "recovering", wait and re-check — do not read the dataset yet. 3. Never track position by timestamp. Persist only the integer `cursor`. If you suspect you missed a commit (a gap between your last known cursor and the current last\_cursor), call: { "op": "reconcile", "fromCursor": \ } This re-emits every manifest you missed, using the same chunk\_ids as the original run. It is always safe to call this — including redundantly. 4. Always upsert on `chunk_id`. Never append blindly and never assume a dataset row is new just because you just saw it — the same chunk\_id appearing twice means "this is still the same fact," not "this changed." 5. Always set maxCostUsd. The actor self-terminates before overspending; it will not silently exceed your ceiling. ```` ### Production Guarantees **Fencing tokens.** Every run that intends to commit must hold the lease `{ run_id, epoch, heartbeat }`. The `epoch` is a fencing token: it increments by exactly 1 every time a lease is stolen from a zombie holder, and every commit re-reads the lease and aborts unless `(epoch, run_id)` still match. This is what makes a hung run safe — if Run A stalls, Run B steals the lease and bumps the epoch, and Run A's late write is rejected rather than silently corrupting state. **Deterministic IDs.** `chunk_id = sha256(source_url + section_path + content_hash)`, joined on a null-byte delimiter specifically so that no combination of URL/path can collide across a field boundary. Same input, same id, every time — across retries, across reconciles, across redeploys. **Upsert-only contract.** Nothing in this system is designed to be appended to. A dataset row is a fact keyed by `chunk_id`; seeing it again is confirmation, not a new event. This is what makes replay (via `reconcile`) free instead of dangerous. ### Architecture 1. **Stage (outbox).** Each chunk is written to the key-value store only, keyed by its deterministic id. Nothing hits the output dataset yet, so a crash mid-run leaves only dead-weight orphans — never a partial commit. 2. **Commit (transaction).** After all chunks are staged, `commit()`: - passes the fencing check, - allocates the next `cursor`, - writes the manifest `{ manifest_id, chunk_ids, cursor, committed_at }`, - emits the referenced chunks to the dataset, - advances `CURSOR_HEAD` — **this HEAD advance is the atomic commit point.** 3. **Reconcile.** `{ "op": "reconcile", "fromCursor": N }` re-emits every committed manifest with `cursor >= N`, reconstructing each chunk from the KV store with its original deterministic id. **Honest trade-off — TOCTOU (Time-of-Check to Time-of-Use).** The Apify key-value store has no compare-and-swap, so lease *acquisition* is a plain read-modify-write and has an inherent TOCTOU window: two runs could both observe a stale lease and both attempt to steal it at nearly the same moment. We do not pretend otherwise. What actually provides safety is the fencing token, checked at the *commit* boundary rather than the *acquisition* boundary — the last writer to steal the lease wins the `run_id` slot, and whichever run loses that race gets rejected the moment it tries to commit, before it can advance `CURSOR_HEAD` or affect a consumer. Acquisition can race; the committed state cannot. True CAS-level single-writer acquisition would require a coordination store that supports it (e.g., a database with real optimistic locking) — we're naming that limitation here on purpose, because a system that's honest about where its safety actually comes from is easier to trust than one that claims perfection. ### Lifecycle example: No-Op → Manifest Commit A consumer polling the health endpoint sees nothing change until a sync actually commits: ```json // Poll 1 — no new data since the consumer's last read. No-op from the // consumer's perspective: last_cursor is identical to what it saw before. { "status": "ok", "staleness_s": 340, "last_cursor": 5 } // A "sync" run executes in between polls, stages 3 chunks, and commits: { "manifestId": "9697071a4e2c...", "cursor": 6, "chunkIds": ["b3c3e00b...", "aa740144...", "e9fb4a91..."], "committedAt": "2026-07-12T09:14:02.331Z", "runId": "run-abc123", "epoch": 1 } // Poll 2 — cursor advanced. The consumer now knows there is exactly one // new manifest (cursor 6) to upsert, and nothing before it needs re-checking. { "status": "ok", "staleness_s": 4, "last_cursor": 6 } ```` ### Health endpoint In Standby mode the actor serves a free-access endpoint at `GET /health`: ```json { "status": "ok" | "recovering", "staleness_s": 42, "last_cursor": 7 } ``` - `staleness_s`: seconds since the last committed manifest (`-1` if none yet). - `status`: `recovering` when nothing has committed yet, or a lease is held with a stale heartbeat. ### Input | Field | Type | Notes | |---|---|---| | `op` | `sync` | `reconcile` | `health` | Default `sync`. | | `sources` | string\[] | http(s) URLs to fetch and chunk (for `sync`). | | `fromCursor` | integer | Re-emit manifests with `cursor >= this` (for `reconcile`). | | `maxCostUsd` | string | **Required.** Hard cost ceiling; the run aborts *before* staging a chunk that would exceed it. String because the Apify form has no float type. | | `costPerChunkUsd` | string | Optional override for the assumed per-chunk cost used to project against `maxCostUsd`. Defaults to 0.0005. | | `baseCostUsd` | string | Optional override for the assumed fixed run cost used to project against `maxCostUsd`. Defaults to 0.01. | | `chunkSizeChars` | integer | Target chunk size when a source has no headings. | | `heartbeatIntervalSecs` | integer | Lease heartbeat cadence; keep well under the 90s zombie threshold. | ### Output (per committed chunk) ```json { "cursor": 1, "manifestId": "9697071a…", "chunkId": "b3c3e00b…", "sourceUrl": "https://…/README.md", "sectionPath": "0:Installation", "contentHash": "…", "committed": true, "runId": "…" } ``` ### Modules - `hasher.ts` — `DeterministicHasher` (content-addressed ids, null-byte-delimited). - `lease-manager.ts` — `LeaseManager` (acquire, heartbeat, zombie-steal, `assertFenced`). - `manifest-manager.ts` — `ManifestManager` (stage, commit, reconcile). - `extractor.ts` — deterministic fetch + sectioning. - `health.ts` — Standby health server. - `retry.ts` — structured backoff that never retries control-flow errors. - `main.ts` — validation, cost ceiling, op dispatch. ### Local development ```bash npm install npm run typecheck npm run harness # in-process invariant tests (deterministic ids, monotonic cursor, reconcile, fencing) npm run build && apify run # real end-to-end sync ``` # Actor input Schema ## `op` (type: `string`): Selects the state-machine verb. 'sync': fetch sources, stage content-addressed chunks, and commit one new manifest at cursor+1. 'reconcile': re-emit every already-committed manifest with cursor >= fromCursor, reconstructing chunks from durable storage with their original deterministic ids — use this to recover a downstream consumer that fell behind or dropped data, never to 'redo' work. 'health': return the {status, staleness\_s, last\_cursor} snapshot and exit without writing anything. ## `sources` (type: `array`): http(s) URLs to fetch and split into deterministic chunks. Required when op = 'sync', ignored otherwise. Each source is sectioned on markdown headings when present, or packed into ~chunkSizeChars windows on paragraph boundaries when not — either way, the same source content always produces the same section boundaries, which is what makes chunk\_id stable across re-runs. ## `fromCursor` (type: `integer`): Used only when op = 'reconcile'. Re-emits every committed manifest whose cursor is >= this value, in cursor order, using the original deterministic chunk\_ids. Set to your last known-good cursor + 1 to backfill exactly the gap a consumer missed — replay is safe because consumers upsert on chunk\_id rather than append. ## `maxCostUsd` (type: `string`): Hard spending ceiling. Checked before every chunk is staged — if internal execution cost projects to exceed this value, the actor self-terminates immediately (CostCeilingError) rather than staging the chunk, so a triggered ceiling never leaves a committed manifest. Prevents runaway billing on unexpectedly large sources. Accepts a decimal string, e.g. '5.00' — string rather than number because the Apify input form has no float type; the actor coerces it internally. ## `costPerChunkUsd` (type: `string`): Override for the assumed marginal cost of staging one chunk, used only to project against maxCostUsd. Defaults to 0.0005 if omitted. Same string-for-float reasoning as maxCostUsd. ## `baseCostUsd` (type: `string`): Override for the assumed fixed cost of a run (container start, initial fetch) before any chunk is staged, used only to project against maxCostUsd. Defaults to 0.01 if omitted. If this alone exceeds maxCostUsd, the run is rejected at validation before any network call is made. ## `chunkSizeChars` (type: `integer`): Target chunk size in characters, used only as a fallback when a source has no markdown headings to section on. Does not affect chunk\_id derivation directly — the resulting section content does, so changing this value after a source has already been synced will produce new chunk\_ids for that source on the next sync. ## `heartbeatIntervalSecs` (type: `integer`): How often the lease heartbeat is refreshed in the KV store while this run holds it. Must stay well under the 90-second zombie-detection threshold (see README) or a live run risks having its own lease stolen out from under it. ## Actor input object example ```json { "op": "sync", "sources": [ "https://raw.githubusercontent.com/apify/crawlee/master/README.md" ], "fromCursor": 0, "maxCostUsd": "5.00", "chunkSizeChars": 1200, "heartbeatIntervalSecs": 20 } ``` # API You can run this Actor programmatically using our API. Below are code examples in JavaScript, Python, and CLI, as well as the OpenAPI specification and MCP server setup. ## JavaScript example ```javascript import { ApifyClient } from 'apify-client'; // Initialize the ApifyClient with your Apify API token // Replace the '' with your token const client = new ApifyClient({ token: '', }); // Prepare Actor input const input = { "sources": [ "https://raw.githubusercontent.com/apify/crawlee/master/README.md" ], "maxCostUsd": "5.00" }; // Run the Actor and wait for it to finish const run = await client.actor("automationnation/delta-sync-sentinel").call(input); // Fetch and print Actor results from the run's dataset (if any) console.log('Results from dataset'); console.log(`💾 Check your data here: https://console.apify.com/storage/datasets/${run.defaultDatasetId}`); const { items } = await client.dataset(run.defaultDatasetId).listItems(); items.forEach((item) => { console.dir(item); }); // 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/js/docs ``` ## Python example ```python from apify_client import ApifyClient # Initialize the ApifyClient with your Apify API token # Replace '' with your token. client = ApifyClient("") # Prepare the Actor input run_input = { "sources": ["https://raw.githubusercontent.com/apify/crawlee/master/README.md"], "maxCostUsd": "5.00", } # Run the Actor and wait for it to finish run = client.actor("automationnation/delta-sync-sentinel").call(run_input=run_input) # Fetch and print Actor results from the run's dataset (if there are any) print("💾 Check your data here: https://console.apify.com/storage/datasets/" + run["defaultDatasetId"]) for item in client.dataset(run["defaultDatasetId"]).iterate_items(): print(item) # 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/python/docs/quick-start ``` ## CLI example ```bash echo '{ "sources": [ "https://raw.githubusercontent.com/apify/crawlee/master/README.md" ], "maxCostUsd": "5.00" }' | apify call automationnation/delta-sync-sentinel --silent --output-dataset ``` ## MCP server setup ```json { "mcpServers": { "apify": { "command": "npx", "args": [ "mcp-remote", "https://mcp.apify.com/?tools=automationnation/delta-sync-sentinel", "--header", "Authorization: Bearer " ] } } } ``` ## OpenAPI specification ```json { "openapi": "3.0.1", "info": { "title": "Delta-Sync Sentinel", "description": "RAG Pipeline, Agentic Infrastructure, Idempotent Ingestion — transactional-outbox delta sync engine with deterministic chunk IDs and lease/fencing tokens. Safe to re-run; manifest-commit is the source of truth.", "version": "0.1", "x-build-id": "nOASlJACAQKheIugl" }, "servers": [ { "url": "https://api.apify.com/v2" } ], "paths": { "/acts/automationnation~delta-sync-sentinel/run-sync-get-dataset-items": { "post": { "operationId": "run-sync-get-dataset-items-automationnation-delta-sync-sentinel", "x-openai-isConsequential": false, "summary": "Executes an Actor, waits for its completion, and returns Actor's dataset items in response.", "tags": [ "Run Actor" ], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/inputSchema" } } } }, "parameters": [ { "name": "token", "in": "query", "required": true, "schema": { "type": "string" }, "description": "Enter your Apify token here" } ], "responses": { "200": { "description": "OK" } } } }, "/acts/automationnation~delta-sync-sentinel/runs": { "post": { "operationId": "runs-sync-automationnation-delta-sync-sentinel", "x-openai-isConsequential": false, "summary": "Executes an Actor and returns information about the initiated run in response.", "tags": [ "Run Actor" ], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/inputSchema" } } } }, "parameters": [ { "name": "token", "in": "query", "required": true, "schema": { "type": "string" }, "description": "Enter your Apify token here" } ], "responses": { "200": { "description": "OK", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/runsResponseSchema" } } } } } } }, "/acts/automationnation~delta-sync-sentinel/run-sync": { "post": { "operationId": "run-sync-automationnation-delta-sync-sentinel", "x-openai-isConsequential": false, "summary": "Executes an Actor, waits for completion, and returns the OUTPUT from Key-value store in response.", "tags": [ "Run Actor" ], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/inputSchema" } } } }, "parameters": [ { "name": "token", "in": "query", "required": true, "schema": { "type": "string" }, "description": "Enter your Apify token here" } ], "responses": { "200": { "description": "OK" } } } } }, "components": { "schemas": { "inputSchema": { "type": "object", "required": [ "maxCostUsd" ], "properties": { "op": { "title": "Operation", "enum": [ "sync", "reconcile", "health" ], "type": "string", "description": "Selects the state-machine verb. 'sync': fetch sources, stage content-addressed chunks, and commit one new manifest at cursor+1. 'reconcile': re-emit every already-committed manifest with cursor >= fromCursor, reconstructing chunks from durable storage with their original deterministic ids — use this to recover a downstream consumer that fell behind or dropped data, never to 'redo' work. 'health': return the {status, staleness_s, last_cursor} snapshot and exit without writing anything.", "default": "sync" }, "sources": { "title": "Source URLs", "type": "array", "description": "http(s) URLs to fetch and split into deterministic chunks. Required when op = 'sync', ignored otherwise. Each source is sectioned on markdown headings when present, or packed into ~chunkSizeChars windows on paragraph boundaries when not — either way, the same source content always produces the same section boundaries, which is what makes chunk_id stable across re-runs.", "items": { "type": "string" } }, "fromCursor": { "title": "From Cursor", "minimum": 0, "type": "integer", "description": "Used only when op = 'reconcile'. Re-emits every committed manifest whose cursor is >= this value, in cursor order, using the original deterministic chunk_ids. Set to your last known-good cursor + 1 to backfill exactly the gap a consumer missed — replay is safe because consumers upsert on chunk_id rather than append.", "default": 0 }, "maxCostUsd": { "title": "Max Cost (USD)", "type": "string", "description": "Hard spending ceiling. Checked before every chunk is staged — if internal execution cost projects to exceed this value, the actor self-terminates immediately (CostCeilingError) rather than staging the chunk, so a triggered ceiling never leaves a committed manifest. Prevents runaway billing on unexpectedly large sources. Accepts a decimal string, e.g. '5.00' — string rather than number because the Apify input form has no float type; the actor coerces it internally." }, "costPerChunkUsd": { "title": "Cost Per Chunk (USD, optional)", "type": "string", "description": "Override for the assumed marginal cost of staging one chunk, used only to project against maxCostUsd. Defaults to 0.0005 if omitted. Same string-for-float reasoning as maxCostUsd." }, "baseCostUsd": { "title": "Base Cost (USD, optional)", "type": "string", "description": "Override for the assumed fixed cost of a run (container start, initial fetch) before any chunk is staged, used only to project against maxCostUsd. Defaults to 0.01 if omitted. If this alone exceeds maxCostUsd, the run is rejected at validation before any network call is made." }, "chunkSizeChars": { "title": "Chunk Size (chars)", "minimum": 200, "type": "integer", "description": "Target chunk size in characters, used only as a fallback when a source has no markdown headings to section on. Does not affect chunk_id derivation directly — the resulting section content does, so changing this value after a source has already been synced will produce new chunk_ids for that source on the next sync.", "default": 1200 }, "heartbeatIntervalSecs": { "title": "Heartbeat Interval (s)", "minimum": 5, "type": "integer", "description": "How often the lease heartbeat is refreshed in the KV store while this run holds it. Must stay well under the 90-second zombie-detection threshold (see README) or a live run risks having its own lease stolen out from under it.", "default": 20 } } }, "runsResponseSchema": { "type": "object", "properties": { "data": { "type": "object", "properties": { "id": { "type": "string" }, "actId": { "type": "string" }, "userId": { "type": "string" }, "startedAt": { "type": "string", "format": "date-time", "example": "2025-01-08T00:00:00.000Z" }, "finishedAt": { "type": "string", "format": "date-time", "example": "2025-01-08T00:00:00.000Z" }, "status": { "type": "string", "example": "READY" }, "meta": { "type": "object", "properties": { "origin": { "type": "string", "example": "API" }, "userAgent": { "type": "string" } } }, "stats": { "type": "object", "properties": { "inputBodyLen": { "type": "integer", "example": 2000 }, "rebootCount": { "type": "integer", "example": 0 }, "restartCount": { "type": "integer", "example": 0 }, "resurrectCount": { "type": "integer", "example": 0 }, "computeUnits": { "type": "integer", "example": 0 } } }, "options": { "type": "object", "properties": { "build": { "type": "string", "example": "latest" }, "timeoutSecs": { "type": "integer", "example": 300 }, "memoryMbytes": { "type": "integer", "example": 1024 }, "diskMbytes": { "type": "integer", "example": 2048 } } }, "buildId": { "type": "string" }, "defaultKeyValueStoreId": { "type": "string" }, "defaultDatasetId": { "type": "string" }, "defaultRequestQueueId": { "type": "string" }, "buildNumber": { "type": "string", "example": "1.0.0" }, "containerUrl": { "type": "string" }, "usage": { "type": "object", "properties": { "ACTOR_COMPUTE_UNITS": { "type": "integer", "example": 0 }, "DATASET_READS": { "type": "integer", "example": 0 }, "DATASET_WRITES": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_READS": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_WRITES": { "type": "integer", "example": 1 }, "KEY_VALUE_STORE_LISTS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_READS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_WRITES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_INTERNAL_GBYTES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_EXTERNAL_GBYTES": { "type": "integer", "example": 0 }, "PROXY_RESIDENTIAL_TRANSFER_GBYTES": { "type": "integer", "example": 0 }, "PROXY_SERPS": { "type": "integer", "example": 0 } } }, "usageTotalUsd": { "type": "number", "example": 0.00005 }, "usageUsd": { "type": "object", "properties": { "ACTOR_COMPUTE_UNITS": { "type": "integer", "example": 0 }, "DATASET_READS": { "type": "integer", "example": 0 }, "DATASET_WRITES": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_READS": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_WRITES": { "type": "number", "example": 0.00005 }, "KEY_VALUE_STORE_LISTS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_READS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_WRITES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_INTERNAL_GBYTES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_EXTERNAL_GBYTES": { "type": "integer", "example": 0 }, "PROXY_RESIDENTIAL_TRANSFER_GBYTES": { "type": "integer", "example": 0 }, "PROXY_SERPS": { "type": "integer", "example": 0 } } } } } } } } } } ```