rfgen.generation

Generation composes configuration, schedules deterministic run/sample/shard seeds, executes locally or remotely, and publishes only through Storage.

RunConfig fields are run_id: str (nonblank), num_samples: int = 10000 (positive), shard_size: int = 1000 (positive), seed: int = 42, and shard_failure_threshold: float = 1.0 in (0, 1]. A partial final shard warns.

ExecutorConfig has name: str = "local", parallelism: int = 1, and dataproc: DataprocServerlessConfig | None. Dataproc settings are required exactly for dataproc_serverless: nonblank project, region, gs:// staging URI, service account, optional image URI, either network or subnet, labels, Spark properties, and extra_packages (local .whl/.zip use-case distributions shipped with the submission). Every extra_packages entry is staged under the name the shared helper staged_package_name computes (rfgen.generation.package_naming), so config-parse-time validation and submission-time staging agree on one rule rather than two.

GenerationConfig(graph, record, storage=None, executor=ExecutorConfig(), run=None) requires record == graph.record. Executor runs require run and storage; Dataproc also requires a gs:// destination. from_hydra(mapping) -> GenerationConfig resolves interpolation and raises ConfigError or Pydantic validation errors for invalid input.

Shipped YAML under rfgen.generation.templates contains authored scenario policy; Python models validate it without selecting scenarios.

GenerateCommand.run(*, params: BaseModel, registry: BaseRegistry[object]) -> int resolves executor and storage plugins. ShardSpec identifies a half-open ordinal interval; build_shard_fn(config, storage_port=...) returns a serializable worker; and publish_run(...) -> RecordCollection validates the complete shard set before one create-only publication. Dataproc workers resolve storage from neutral configuration and never construct SDS directly.

SeedSchedule derives run, shard, and sample generators. Structural graph draws remain in rfgen.graph.randomness, preserving dependency direction.

Executor extension contracts

BaseDistributedExecutor(ABC) implements the local/distributed callable boundary. Subclasses implement submit(shard_fn, shards) -> ExecutorRunHandle, wait(handle) -> RunResult, and cancel(handle) -> None. A handle is owned by the instance that created it; passing it to another instance raises ValueError. _result and _require_handle are sealed normalization helpers.

RemoteGenerationExecutor(ABC) is for environments that cannot serialize a Python shard callable. It requires from_generation_config(config), submit_remote(shards), wait(handle), cancel(handle), and lifecycle_commands(remote_job, result_uri=None). Remote workers reconstruct the complete validated configuration and storage selector. Register either executor under rfgen.executors; LocalExecutor and DataprocServerlessExecutor are the built-in concrete examples.

For an operator-controlled remote launch, DataprocLaunchPlan freezes the prechosen native batch ID, staging root, every staged DataprocArtifactIdentity, and the complete resolved command before billing. DataprocLaunchAuthority is the public create/consume ABC; its contract includes exact create, state, and one-use consume operations. DataprocLaunchAuthorization carries storage-neutral authority identity and revision fields. DataprocPreparedApplicationExecutor receives this interface and never constructs a backend. Stock Dataproc wiring supplies GcsDataprocLaunchAuthority, which stores the authorization as a create-only GCS object and consumes its exact generation once; custom implementations need not use GCS. Expiration timestamps remain exact, including microseconds, because they are part of the authorized bytes. Preparation is side-effectful staging but not submission. DataprocPreparedApplicationExecutor then consumes authority and asks GcsDataprocObjectCustody to server-side-copy each authorized source generation to a fresh execution URI. The copy binds the source generation and destination nonexistence. Before the first copy, a create-only PREPARING manifest records a unique custody token and every expected source and execution identity. The rewrite request atomically creates the destination with a temporary hold plus the custody token and source- generation metadata. DataprocMaterializedArtifact records the held generation. The hold protects lifecycle mistakes and deletion during provisioning; it does not defend against a storage administrator who can release holds. After each rewrite, a generation-conditional manifest update records progress. If a launcher stops between rewrite and that update, recovery examines only the known execution URI and adopts it only when the hold, custody token, and source generation all match. A mismatch stops and is never deleted as if it were owned. The public recover operation starts only from READY, CREATE_ATTEMPTED, SUBMITTED, or ABANDONED. A PREPARING manifest refuses without writes or a remote call; the sole supported continuation is submit with the already consumed authorization, which resumes the conditionally recorded artifact prefix.

Only a generation-conditional READY transition permits asynchronous batch creation. Before the sole create RPC, core durably advances the manifest to CREATE_ATTEMPTED. A READY recovery may perform that transition and create once; a CREATE_ATTEMPTED recovery never recreates a batch because Dataproc IDs can be reused after deletion. It reconciles the prechosen resource after bounded describe attempts. Exact effective PySpark fields advance to SUBMITTED; an absent batch conditionally advances the same manifest generation to ABANDONED, then writes the abandon receipt and revokes held inputs. SUBMITTED and ABANDONED therefore cannot both win, including under concurrent recovery. Recovery from ABANDONED resumes the exact receipt and cleanup only; a mismatch stops without cleanup. describe_retry_delay_s configures the positive delay between bounded native describe attempts; tests may inject a wait callable without sleeping.

Dataproc describes effective configuration rather than literal CLI arguments. Reconciliation checks the resource name, UUID, creation time, custody-token and other user labels, driver URI, arguments, ordered Python files, image, service account, network or subnetwork, staging bucket, and Spark properties. It accepts documented service additions (goog-dataproc-*, application identity and spark.dataproc.*) and canonical Compute Engine URI expansion; other drift is a failure. DataprocPreparedRun carries the manifest’s native identity for crash recovery. Once the exact batch is terminal, finalization writes a create-only DataprocFinalizationReceipt, then releases and deletes only held execution inputs—never results. Exact receipt replay resumes partial cleanup; a mismatch stops. Authoritative batch absence uses the same decision URI to abandon an unsubmitted plan, so abandon and finalize cannot both win. Authorization is optional for the ordinary stock path. Dataproc’s unique batch resource rejects concurrent duplicate submission, while durable authority prevents replay after that batch is deleted.

Finalization re-reads the current manifest and accepts only its exact SUBMITTED generation. It compares the current batch’s immutable effective identity, including UUID and creation time, with the persisted submission evidence before writing a receipt or cleaning inputs. Deleting and recreating a same-ID, same-configuration batch cannot inherit the earlier submission.