Sync workflows

Best practices for ETL workflows that keep a Brain in sync with an external app.

A sync workflow is an ETL pipeline: it discovers what changed in a source app, transforms each item into a document, and upserts it into a Brain — while also removing documents whose source items are gone. This page is the playbook for building one that is correct, fast, and cheap to run repeatedly.

Step 0 — discover what the source API offers

Before writing any sync code, research the app’s API and classify its change-detection support. This decision drives the whole workflow structure, so do it first — check the app’s API docs (and the integration’s available actions) for each of these, in order:

Tier	Capability to look for	Examples of what it’s called
1. Change feed / delta cursor	An opaque token you persist; the API returns everything that changed since — including deletions — plus a new token	”changes API”, “delta query”, “events stream”, “history API”, “cursor”
2. Updated-at queries	List/search endpoints that filter or sort by modification time, so you can fetch only items changed after a high-water mark	updated_since, lastModified >, last_edited_time filters, updatedAt sort
3. Webhooks / events	The app pushes “item X changed” notifications to you	”webhooks”, “event subscriptions”, “notifications”
4. Full enumeration only	You can list everything, and nothing else	plain paginated list endpoints

Most apps offer several. Note two things for each capability you find: does it report deletions? (most timestamp queries do not; most change feeds do) and is it exhaustive? (some search endpoints are explicitly not guaranteed to return every item — check the fine print).

The layered structure

Production sync systems never rely on a single mechanism. Build up to three layers, and let each one own what it is good at:

1. Initial full sync — enumerate everything once, upload it all. For feed-based sources, capture the change cursor before the scan starts, so changes made mid-scan replay through the feed afterwards with no gap.
2. Incremental fast path — the cheap, frequent layer: consume the change feed, query by updated-at high-water mark, or react to a webhook trigger. Runs on a schedule (or on events) and touches only what changed.
3. Reconciliation sweep — a periodic full enumeration + diff (for example nightly or weekly) that catches everything the fast path missed: dropped webhooks, non-exhaustive search indexes, and deletions that timestamp queries can’t see.

If the source has a tier-1 change feed, layers 2 and 3 collapse into one: the feed is both fast and complete (it reports deletions), so a full scan is only needed once. Otherwise, treat the fast path as an optimization and the sweep as the source of truth.

Webhooks are hints, not truth. Deliveries are typically batched, reordered, and at-most-once — a webhook-only sync silently drifts. Use webhooks to trigger a targeted upsert quickly, and keep a scheduled sweep that owns correctness.

Recommended structure by tier

Tier 1 — change feed (best case). Two phases in one workflow: a full phase that pages through the entire corpus once, then an incremental phase that consumes the feed forever. Persist the phase + cursor between runs. Deletions arrive in the feed.

Tier 2 — updated-at queries. Keep a high-water mark (the max modification time you’ve processed). Each run fetches items newer than the mark, upserts them, and advances the mark. Add a scheduled reconciliation sweep for deletions — a plain full listing diffed against the Brain.

Tier 3 — webhooks. Wire the webhook to a trigger that runs a small workflow upserting (or deleting) the single referenced item. Keep the scheduled sweep.

Tier 4 — full enumeration. Every run lists everything and diffs against the Brain. This is self-healing by construction and perfectly fine for corpora up to a few thousand items — unchanged items short-circuit without uploading (see below), so re-runs are cheap.

Idempotency: documentIds and revisions

Make every run a pure upsert so re-running is always safe:

• Stable documentIds — derive from the source’s immutable identifier, prefixed by source: linear:{issueId}, notion:{pageId}, google-drive:{fileId}. The prefix lets you list({ prefix }) just your documents and keeps sources from colliding.
• revision is your change fingerprint — pass the source’s version marker (updatedAt, last_edited_time, a version number, or an ETag). Compare it before fetching content: list existing documents first, skip items whose stored revision already matches, and only download/transform/upload the rest. The platform also content-hashes uploads server-side, so an upload with identical content resolves as outcome: "unchanged" rather than re-embedding — a second safety net, not a replacement for the revision check (skipping locally avoids the fetch entirely).

// Load the Brain's view of this source once per run — it is your sync ledger.
const existing = new Map<string, string | null>();
let cursor: string | undefined;
do {
  const page = await knowledge.list({ prefix: "linear:", cursor, limit: 500 });
  for (const doc of page.documents) existing.set(doc.documentId, doc.revision);
  cursor = page.nextCursor ?? undefined;
} while (cursor);

// Later, per source item:
if (existing.get(documentId) === item.updatedAt) {
  unchanged += 1;
  continue; // no fetch, no upload
}

Deletion handling

A sync that never deletes serves stale answers forever. Match the mechanism to your tier:

• Change feed: apply the feed’s removal events with deleteMany.
• Everything else: during a full enumeration, track every documentId you saw; anything in the Brain (under your prefix) that you didn’t see is a candidate for deletion. If the listing API is not guaranteed exhaustive, recheck each unseen item individually (fetch it by id) before deleting — a missing item deletes, a present one stays.
• Delete in batches (deleteMany accepts up to 100 ids) rather than one at a time.

Durability and parallelism

Sync workflows run on Keystroke’s durable workflow engine — every action call and Brain operation is a recorded step, so a crashed or retried run resumes after its last completed step instead of starting over. That has two practical consequences:

• Prefer one long run over many small resumable runs. You don’t need to checkpoint after every page for safety; replay already covers crashes. Chained short runs are only worth it when a single run would exceed practical limits (tens of thousands of items).
• Parallelize the per-item work, serially walk the cursors. Pagination is inherently sequential (each page token depends on the last), but fetching content and uploading documents is not. Collect changed items while paging, then process them with a bounded worker pool — around 5–8 concurrent uploads is a good default; ingestion is compute-bound, so more rarely helps.

// Bounded pool: workers pull from a shared queue until it drains.
async function runWithConcurrency<T>(
  items: readonly T[],
  concurrency: number,
  fn: (item: T) => Promise<void>,
): Promise<void> {
  let next = 0;
  await Promise.all(
    Array.from({ length: Math.min(concurrency, items.length) }, async () => {
      while (next < items.length) await fn(items[next++]!);
    }),
  );
}

await runWithConcurrency(changed, 5, async (item) => {
  try {
    await knowledge.uploadDocument({ /* … */ });
    uploaded += 1;
  } catch (error) {
    failed += 1; // count and report; never abort the whole sync for one item
  }
});

Keep the worker callback — the code that calls .run() and uploadDocument() — in the workflow file itself, per the authoring best practices: steps called from imported helpers lose durability and canvas attribution. A pooled section renders as a single opaque block on the workflow canvas (the build warns about this); that is a cosmetic trade-off, and the steps inside remain fully durable.

Catch per-item failures inside the pool and report them in the workflow output (counts plus a capped error list) — one unexportable file should not fail a 500-document sync. Reserve thrown errors for structural problems (a broken cursor, an inconsistent API response) where continuing would corrupt state.

Storing sync state

Tier-2 and tier-1 syncs need a small piece of persistent state (a high-water mark or cursor) between runs:

• The simplest durable home is the Brain itself: a document under a reserved prefix such as system:{source}-sync-state, with source: "system". Write the new state at the end of a successful run — never mid-run — so a failed run retries from the previous state.
• Cursors and tokens are account-bound. If the connected credential changes accounts, delete the state document (and usually the source’s documents) so the next run performs a fresh full sync — an old cursor silently reports the wrong account’s changes.
• Tier-4 syncs need no state at all: the Brain’s stored revisions are the state.

Output shape

Return counters so runs are observable at a glance and comparable over time:

output: z.object({
  discovered: z.number().int(),
  uploaded: z.number().int(),
  unchanged: z.number().int(),
  deleted: z.number().int(),
  failed: z.number().int(),
  errors: z.array(z.string()), // capped, human-readable
}),

A healthy steady-state run looks like discovered: N, unchanged: N, uploaded: 0 — if a no-change run uploads documents, your revision fingerprint is unstable (a common cause: deriving revision from data that changes on every fetch).

Checklist

• Researched the source API: change feed? updated-at filters? webhooks? Does anything report deletions? Is listing exhaustive?
• documentIds are {source}:{stableExternalId}; every upload sets source and revision
• Unchanged items short-circuit from the stored revision, before fetching content
• Deletions handled (feed events, or seen-set diff with recheck)
• Per-item work pooled (5–8), per-item failures counted rather than thrown
• Sync state (if any) written once, at the end of a successful run
• A scheduled trigger runs the fast path; a sweep owns correctness when the fast path can miss changes