Great question. Assuming Prefect 3.x, here’s what works well for watermarking/high‑water marks and state management, plus a scalable pattern you can adopt. Useful Prefect features for watermarks/state - Variables (recommended): Durable, globally accessible key/value storage for cursors like timestamps/IDs. Docs: Variables - Result persistence + caching: Persist task results and cache by inputs to make runs resumable/efficient; back results with S3/GCS/etc. Docs: Results and Docs: Caching - Concurrency limits: Prevent concurrent updates to the same watermark to avoid races; applied by tag. Docs: Concurrency limits - Artifacts: Record an audit trail of watermark changes and progress. Docs: Artifacts - Retries and idempotency: Use task retries + idempotent loaders so watermarks only advance after a successful commit. Docs: Retries - Events (optional): Emit custom events when you advance a watermark for downstream automation. Docs: Events - External blocks/integrations (optional): Keep watermarks in Redis/Postgres/cloud stores if you need org‑wide shared state. Recommended scalable pattern - Store the cursor - Default: Prefect Variables with a clear naming scheme per source/partition (e.g.,

wm:orders:partition=12

). - If your platform standardizes on an external store, use a Prefect block/integration for that store. - Process flow 1) Read the watermark at the start of the extract. 2) Process records starting at that watermark. 3) Only after a successful load/commit, update the watermark to the next value. 4) Write an artifact/audit record and optionally emit an event. - Concurrency - Tag the task that updates the watermark and create a concurrency limit of 1 for that tag to avoid concurrent updates. - For parallelism, shard by partition (distinct variable keys and tags per partition) so each shard can progress independently. - Resilience/observability - Enable

persist_result=True

on tasks/flows and configure

result_storage

(e.g., S3/GCS) for durable checkpoints. - Use retries on networked steps and idempotent loaders to avoid duplicate writes. Minimal code pattern (single partition)

from prefect import flow, task
from prefect.variables import Variable

@task(tags=["wm:orders"])  # tag used for concurrency limit
def extract_from_watermark(batch_size: int = 1000):
    start_id = int(Variable.get("wm:orders", default=0))
    end_id = start_id + batch_size
    # fetch from source [start_id, end_id)
    return {"start_id": start_id, "end_id": end_id, "records": list(range(start_id, end_id))}

@task
def transform(batch):
    # do transformations
    return batch

@task
def load_then_advance(batch):
    # load records atomically in your destination
    # only after success, advance the watermark
    Variable.set("wm:orders", batch["end_id"], overwrite=True)
    return batch["end_id"]

@flow(persist_result=True, log_prints=True)
def orders_pipeline(num_batches: int = 5):
    for _ in range(num_batches):
        batch = extract_from_watermark(batch_size=1000)
        batch = transform(batch)
        new_wm = load_then_advance(batch)
        print(f"Advanced watermark to {new_wm}")

Apply a concurrency limit (verified CLI)

prefect concurrency-limit create "wm:orders" 1

Partitioned pattern (scales horizontally) ``` from prefect import flow, task from prefect.variables import Variable def wm_key(source: str, partition: int) -> str: return f"wm{source}partition={partition}" def wm_tag(source: str, partition: int) -> str: return f"wm{source}partition={partition}" @task

def extract(source: str, partition: int, batch_size: int = 1000): key = wm_key(source, partition) start = int(Variable.get(key, default=0)) end = start + batch_size return {"key": key, "tag": wm_tag(source, partition), "start": start, "end": end, "records": list(range(start, end))} @task def transform(batch): return batch @task def load_then_advance(batch): # load Variable.set(batch["key"], batch["end"], overwrite=True) return batch["end"] @flow(persist_result=True) def partitioned_pipeline(source: str, partitions: list[int]): for p in partitions: b = extract(source, p, batch_size=1000).with_options(tags=[wm_tag(source, p)]) b = transform(b) load_then_advance(b).with_options(tags=[wm_tag(source, p)])

Add per-partition concurrency limits (verified CLI)

# Example for one partition tag; script to create for all partitions if needed prefect concurrency-limit create "wmorderspartition=0" 1

Optional: result persistence on tasks to back with cloud storage

from prefect import task from prefect_aws import S3Bucket, AwsCredentials s3 = S3Bucket(bucket_name="my-checkpoints", credentials=AwsCredentials(...)) s3.save("my-checkpoints") @task(persist_result=True, result_storage=s3) def heavy_step(...): ...

Optional: artifacts for audit

from prefect.artifacts import create_table_artifact create_table_artifact( key="etl-watermarks", table=[{"source": "orders", "new_watermark": 12345}], description="Watermark changes" ) ``` Links - Variables: Variables - Results: Results and persistence - Caching: Caching - Concurrency limits: Concurrency limits - Artifacts: Artifacts - Events: Events If you share your source system and partitioning scheme, I can tailor the variable keys, concurrency tags, and result storage to your setup.

Short answer: your plan is basically right. A couple of refinements will make it robust and scalable. Is this the right approach? - Yes: 1) Upstream task returns list of ids. 2) Persist that task’s result (local or GCS) so downstream can resume if needed. 3) Only after that task succeeds, advance a watermark for that logical stage. 4) Downstream reads the persisted results, does more work, writes to GCP. 5) Only after successful write, advance the downstream stage’s watermark. Where to keep watermarks and how to name them - Store watermarks in Prefect Variables (they’re global per workspace). You don’t have to tie them to a single “task” identity; instead, scope by the logical data boundary the watermark represents (e.g., a source/table/partition). - Naming convention to avoid conflicts (include env, flow, stage, source, and partition/shard): - Example key:

wm:{env}:{flow}:{stage}:{source}[:{partition}]

- Examples: -

wm:prod:orders_flow:extract:orders_table

-

wm:prod:orders_flow:load:orders_table:region=us

- If multiple parallel tasks may update the same logical watermark, tag those tasks and enforce a single-writer with a concurrency limit of 1 for that tag: - Command (verified):

prefect concurrency-limit create "wm:prod:orders_flow:extract:orders_table" 1

- Docs: Concurrency limits Minimal pattern

from prefect import flow, task
from prefect.variables import Variable

def wm_key(env: str, flow: str, stage: str, source: str, partition: str | None = None) -> str:
    return f"wm:{env}:{flow}:{stage}:{source}" + (f":{partition}" if partition else "")

@task(persist_result=True, tags=["wm:prod:orders_flow:extract:orders_table"])
def produce_ids(...):
    # fetch ids since watermark
    key = wm_key("prod", "orders_flow", "extract", "orders_table")
    start = int(Variable.get(key, default=0))
    ids = fetch_ids_since(start)  # your logic
    return {"ids": ids, "new_wm": max(ids) if ids else start}

@task
def consume_and_write(batch):
    write_to_gcp(batch["ids"])
    return batch["new_wm"]

@task
def advance_watermark(key: str, value: int):
    Variable.set(key, value, overwrite=True)

@flow(persist_result=True)
def pipeline():
    key = wm_key("prod", "orders_flow", "extract", "orders_table")
    batch = produce_ids()
    new_wm = consume_and_write(batch)
    # advance AFTER successful write
    advance_watermark(key, new_wm)

Result persistence behavior (Prefect 3.x) - When are results persisted? - Only if you enable it (e.g.,

@task(persist_result=True)

or set a cache policy or explicit `result_storage_key`/`result_storage`). Default is not persisted. Docs: Results - Do results last forever? - Prefect does not auto-delete persisted results. You manage retention: - Local: delete files from

~/.prefect/storage/

(or your configured path). - GCS/S3: use lifecycle rules to auto-expire or delete by prefix. -

cache_expiration

prevents cache reuse after a time window but does not delete stored objects. Docs: Caching - Are results “partitioned”? - They’re stored by a storage key. You control the “partitioning” by templating the key and/or choosing the cache policy. - Use

result_storage_key

to layout objects in storage (e.g., include flow, task, partition, and schedule date).

@task(
      persist_result=True,
      result_storage_key="results/{flow_run.name}/{task_run.name}/{parameters[partition]}/{prefect.runtime.flow_run.id}.pkl"
    )

- For caching, the default cache policy includes the run id (DEFAULT = INPUTS + TASK_SOURCE + RUN_ID), which means cache is not reused across different flow runs by default. To reuse across runs, choose a cache policy like

INPUTS

or custom. Reruns and failed runs - If a previous run failed:

- Any successful tasks that persisted results will still have objects in storage. They’re only reused on rerun if your cache policy yields the same cache key and the cache has not expired. - If you want to re-execute even when a cache exists, set

refresh_cache=True

on the task or globally: - Per task:

@task(cache_policy=INPUTS, refresh_cache=True, persist_result=True)

- Global:

prefect config set PREFECT_TASKS_REFRESH_CACHE=true

- If you want to completely isolate results by run: - Include

prefect.runtime.flow_run.id

in your

result_storage_key

so reruns don’t touch prior run artifacts. - If you want to reuse results across runs: - Exclude run id from the storage key and use an

INPUTS

cache policy; optionally set a

cache_expiration

. Cleanup options - Local storage: delete the files by path/prefix. - GCS/S3: use bucket lifecycle policies or scripted deletes by prefix. - Note: deleting objects removes the ability to hydrate cached states that reference them; this is fine if you don’t need old caches. Best practices for your case - Use one watermark variable per logical data boundary (source/partition/stage), not per arbitrary task function. - Update watermarks only after successful commit/write. - Add concurrency limits for tasks that update the same watermark to avoid race conditions. - Choose a

result_storage_key

layout that fits your retention policy and rerun strategy: - Per-run isolation:

results/{env}/{flow}/{stage}/{task}/{parameters[partition]}/{prefect.runtime.flow_run.id}.pkl

- Cross-run reuse:

results/{env}/{flow}/{stage}/{task}/{parameters[partition]}.pkl

- Configure storage lifecycle rules (GCS/S3) for retention. Helpful links - Variables: Variables - Results and persistence: Results - Caching: Caching - Concurrency limits: Concurrency limits - Events (optional): Events If you share your source(s), partitioning scheme, and retention goals, I can propose concrete variable keys, result_storage_key templates, and bucket lifecycle rules.