Cloud Backends¶

rfgen keeps cloud SDKs out of its core import path. Cloud storage, credentials, and managed batch annotation use optional extras; distributed generation uses PySpark through the shipped executor registry.

For environment-specific setup, see Reference / Compute Environment.

Execution model¶

LocalExecutor runs synchronously on one host. PySpark is the only shipped distributed execution model. Dask and Ray are deferred and have no executor classes, extras, or registry selectors.

The shipped rfgen.executors entry points are:

Selector	Class	Execution environment
`local`	LocalExecutor	Synchronous local process
`spark`	SparkExecutor	User-supplied or ambient Spark cluster
`gcp_serverless_spark`	ManagedSparkServerlessExecutor	Google Cloud Spark environment
`emr`	EMRExecutor	Amazon EMR Spark environment
`synapse`	SynapseExecutor	Azure Synapse Spark environment
`databricks`	DatabricksExecutor	Databricks Spark environment

The four managed classes inherit the shipped SparkExecutor behavior. They do not import or construct vendor SDK clients. The application host supplies a configured SparkSession, a zero-argument session factory, or an ambient SparkSession.builder configuration.

PySpark is not installed by a cloud extra. Install and configure a PySpark version compatible with the target Spark service in the driver and worker environment.

Managed-Spark control protocol¶

Layer 19 ships the provider-neutral control-plane contract in rfgen.managed_spark; it does not ship a provider SDK adapter or submit a job to a cloud service. A deployment supplies a ManagedSparkAdapter for one of gcp_dataproc_serverless, aws_emr_serverless, azure_synapse, or databricks. The adapter owns provider authentication and transport; the core package never imports a cloud SDK at this boundary.

ManagedSparkJobSpecV1 is a strict, immutable canonical request. It binds a run UUID, wheel URI and digest, entry point, arguments, labels, identity and network maps, an absolute UTC deadline, and spec_sha256. Secrets must remain in IAM or the adapter configuration, never in the spec maps or logs. Submit requires a non-empty idempotency key. Reusing it with the same request returns the original handle; reusing it with another request is an IdempotencyConflictError.

Every handle, status, result, log page, and reconciliation report carries a canonical SHA-256 self-hash. Status uses the normalized lifecycle SUBMITTED, QUEUED, RUNNING, CANCELLING, SUCCEEDED, FAILED, CANCELLED, or UNKNOWN. An adapter must reconcile UNKNOWN before waiting or cancelling can proceed; an unresolved outcome is a protocol error rather than permission to create a second job. Provider request timeouts are fixed at 30 seconds in the request contract. Cancellation must be given a UTC deadline no more than five minutes ahead; waits are bounded to seven days.

Log pagination is handle-bound and opaque, with a limit from 1 through 1,000. Reconciliation returns immutable rows describing the state observed, action, state after action, and evidence digest. The bundled InMemoryManagedSparkAdapter is deterministic reference/test support only; it is not a cloud emulator or production backend.

Executor configuration¶

The executor config surface has two fields:

Field	Type	Default	Behavior
`name`	`str`	`local`	Open entry-point selector. Blank names are rejected.
`parallelism`	positive `int`	`1`	Requested PySpark partition count. Ignored by synchronous local execution.

Free-form executor parameters are rejected. Service-specific Spark settings, authentication, networking, and resource sizing belong in the supplied SparkSession configuration.

executor:
  name: gcp_serverless_spark
  parallelism: 32

Equivalent selectors for the other shipped paths are local, spark, emr, synapse, and databricks.

Runtime behavior¶

Every executor accepts a one-shot iterable of shard specifications. Before dispatch, it calls the injected store’s has_shard(shard_id) completion probe and skips completed shards. Store construction and record writes remain in the storage and shard-worker boundaries.

SparkExecutor:

plans pending shards in chunks of 10_000;
creates one flat SparkContext.union when multiple chunks are present;
applies the supplied shard function as a Spark transformation;
streams terminal results through toLocalIterator();
assigns a Spark job group so cancel(handle) can call cancelJobGroup(...); and
retains only the configured result-detail bound in public handles and results.

Importing rfgen.executors does not import PySpark or a cloud SDK. PySpark is imported only if a Spark session was not injected and the executor must use SparkSession.builder.getOrCreate().

Cloud extras¶

Cloud extras provide credential, storage, inference, and batch-annotation SDKs. They are not alternate distributed executor implementations.

Extra	SDK dependencies	Shipped capabilities
`rfgen[gcp]`	Google Cloud Storage, Google Auth, Google Gen AI	GCP credentials, GCS storage, Gemini inference, Vertex batch annotation
`rfgen[aws]`	boto3	AWS credentials, S3 storage, SageMaker batch annotation
`rfgen[azure]`	Azure Identity, Azure Blob Storage	Azure credentials, Blob storage, Azure batch annotation
`rfgen[openai-batch]`	OpenAI SDK	OpenAI batch annotation
`rfgen[anthropic-batch]`	Anthropic SDK	Anthropic batch annotation

Combine extras when a workflow needs more than one provider:

pip install 'rfgen[gcp,anthropic-batch]'

Provider clients and transports lazy-import their official SDKs. Missing SDKs raise BackendUnavailableError when the provider path is constructed, not when the core package is imported.

Offline admission estimates¶

Admission estimation consumes immutable price and quota snapshots. It does not call provider pricing or quota APIs during deterministic replay. The caller supplies a configured Layer 14 SigstoreDSSEAdapter as the required estimate(..., adapter=...) verifier. The estimator calls only its verify method, with the PRICING_V1 policy, pricing-bundle.json subject, and the strict PricingBundleV1 predicate model. Tests may supply a structurally compatible verifier fake; production must supply the Layer 14 adapter.

Known object-store schemes bind to their provider in billable plans: gs to GCP, s3 to AWS, and az, abfs, abfss, or azure to Azure. Local storage is explicitly nonbillable and never enters provider pricing.

Storage retention is billed in GB-months, with one month fixed at 730 hours. Plans may retain a month interval directly. When they also retain source hours, each month endpoint must exactly equal its hours endpoint divided by 730; the admission mapper uses the derived interval and records the hours source.

Quota enforcement¶

Quota enforcement consumes the already-complete offline estimate and does not re-query provider pricing or quota APIs. It rejects incomplete estimates, missing or extra unit ceilings, and hard-limit breaches; it admits a request inside every soft limit without an approval; and it requires a signed approval only in the range between soft and hard limits. A supplied approval is verified through the injected Layer 14 DSSE adapter, bound to the exact budget, estimate, resource spec, signer, and policy, then single-used through a Redis replay claim. Any verification or Redis uncertainty rejects the decision rather than admitting it.

The optional operational guards are separate from that decision: limits owns Redis fixed-window rate-limit TTLs, PostgreSQL advisory locks bound concurrency, and a caller-provided run-control adapter must report a PAUSING state observed no later than 30 seconds after a hard breach. These contracts do not provision a Redis, PostgreSQL, or run-control deployment. See the Quota Enforcement API for exact models, signatures, errors, and failure precedence.

Transactional publication¶

The storage transaction boundary publishes a new dataset revision in two steps. It first writes every staged object to a unique immutable key, then it conditionally advances the dataset LATEST pointer. Readers use only the revision named by that pointer and verify each referenced object checksum. A failed publication leaves the prior pointer and its visible objects intact.

The provider adapters use official SDK conditional-write APIs. The cloud SDK clients are supplied to the adapters, so importing the core package does not construct provider clients.

Provider	Immutable-object condition	`LATEST` pointer condition	Minimum dependency
Amazon S3	`put_object(..., IfNoneMatch="*")`	`put_object(..., IfMatch=observed_etag)`	`boto3>=1.36`
Google Cloud Storage	`Blob.upload_from_string(..., if_generation_match=0)`	`Blob.upload_from_string(..., if_generation_match=observed_generation)`	`google-cloud-storage>=2.18`
Azure Blob Storage	`BlobClient.upload_blob(..., overwrite=False)`	`BlobClient.upload_blob(..., etag=observed_etag, match_condition=MatchConditions.IfNotModified)`	`azure-storage-blob>=12.24`

S3 multipart uploads do not carry these write preconditions. A multipart transfer therefore completes to a unique immutable key before the pointer compare-and-swap operation. Azure lease operations use BlobLeaseClient in addition to the ETag condition on the pointer.

fsspec is a read and list transport only. It never performs immutable writes, pointer compare-and-swap, or lease operations because it does not offer a portable conditional-write contract.

Cloud recovery inventory and quarantine are not defined by this page. A deployment that needs either capability supplies provider-specific operational procedures; it must not treat a generic listing result as authorization to delete or quarantine cloud objects.

Layer 43: disaster-recovery release evidence¶

Treat a provider recovery exercise as release evidence only when its retained, immutable bundle names the provider, account or project, region, backend and runbook revisions, target dataset revision, and UTC start and end times. Preserve the conditional-write observations, recovery inventory and reconciliation report hashes, restored-manifest verification report, measured RPO and RTO, and the exercise outcome with its approving operator. Provider logs alone are insufficient: redact secrets, retain the relevant exported records, and mark partial, failed, or unverified exercises as non-passing.

Register a third-party executor¶

Third-party packages register an executor class through the same entry-point group:

[project.entry-points."rfgen.executors"]
custom_spark = "rfgen_custom:CustomSparkExecutor"

The class implements the BaseDistributedExecutor surface:

from collections.abc import Iterable

from rfgen.executors import BaseDistributedExecutor, ExecutorRunHandle
from rfgen.orchestration import RunResult, ShardResult, ShardSpec


class CustomSparkExecutor(BaseDistributedExecutor):
    def submit(
        self,
        shard_fn,
        shards: Iterable[ShardSpec],
    ) -> ExecutorRunHandle:
        ...

    def wait(self, handle: ExecutorRunHandle) -> RunResult:
        ...

    def cancel(self, handle: ExecutorRunHandle) -> None:
        ...

The registry selector remains an open string, so installing the package is enough to make executor.name: custom_spark discoverable. A core framework change is not required.

Admission estimation¶

rfgen.admission_estimation constructs an offline, canonical cost-admission artifact for one already-resolved run. It does not call cloud pricing or quota APIs, select a cloud vendor, or create resources. The caller supplies the signed pricing bundle and the short-lived provider quota snapshot that it has already obtained.

All V1 models in this surface are immutable, strict Pydantic models. The models that declare a *_sha256 self-hash validate it as a SHA-256 digest over RFC 8785 canonical JSON; embedded plan, demand, provenance, and line-item models without such a field do not self-hash. Callers must not mutate any model through model_construct or substitute equivalent-looking unordered data. See the Admission Estimation API for the complete field-level contract.

Models by purpose¶

Purpose	Public models and enums	Contract
Price evidence	`TierV1`, `PriceSnapshotV1`, `FXSnapshotV1`, `PricingBundleV1`, `CloudProvider`, `PriceUnit`	A bundle has uniquely sorted provider/SKU/region/unit prices, optional EUR FX rows, UTC creation and expiry times, and a canonical bundle hash. A price row has contiguous tiers beginning at zero.
Admission plan	`QuantityIntervalV1`, `ExecutorAdmissionPlanV1`, `AnnotationAdmissionPlanV1`, `StorageAdmissionPlanV1`, `AdmissionPlanningConfigV1`, `StorageBillingMode`, `LocalNonbillableReason`	Every quantity is a finite `low <= expected <= high` interval. The plan names exact provider, region, SKU, and quota identities; it is not a provider-SKU inference mechanism.
Provenance and resource spec	`ResolvedPluginV1`, `PluginLockV1`, `ResolvedConfigProvenanceV1`, `AdmissionResourceSpecV1`	The spec binds the resolved run, plan, exact plugin wheels, resolved configuration hashes, source Git revision, and UTC resolution time.
Mapped resources	`ResourceDemandV1`, `NonbillableResourceV1`	Mapping produces sorted billable provider-qualified demands and retains local storage as a nonbillable provenance row.
Quota evidence	`QuotaRowV1`, `QuotaSnapshotV1`, `QuotaUnit`, `QuotaValidationCode`	A quota snapshot contains uniquely sorted exact provider-native readings and expires no more than one hour after creation. Unit conversion is forbidden.
Estimate result	`CostLineItemV1`, `UnitCostTotalV1`, `MissingCostInputV1`, `CostEstimateV1`, `MissingInputReason`, `CostEstimateStatus`	Results retain source digests, interval endpoints, typed missing inputs, exact per-unit totals, currency totals, and a canonical estimate hash. `unit_total_projection(...)` is the fixed five-field projection used to hash each unit total.

Construct the resource spec¶

The single configuration boundary is:

build_admission_resource_spec(
    resolved_cfg: DictConfig,
    planning: AdmissionPlanningConfigV1,
    *,
    plugin_lock: PluginLockV1,
    source_revision: str,
    resolved_at: datetime,
) -> AdmissionResourceSpecV1

resolved_cfg must be an OmegaConf DictConfig, not a pre-resolved dict. The constructor resolves every interpolation with missing-value errors enabled, then strictly validates the generation, executor, storage, and optional annotation contracts. It rejects mismatched record counts, executor selectors, storage selectors, storage schemes, and annotation selectors with AdmissionInputError(code="CONFIG_MISMATCH").

The supplied plugin lock must name uniquely sorted (group, name) selectors, their distribution/version/wheel SHA-256 identities, and its own canonical hash. A malformed or unresolved configuration, invalid lock, invalid source revision, or invalid provenance construction raises AdmissionInputError(code="PROVENANCE").

Storage provider and retention rules¶

For billable storage, gs binds to GCP, s3 to AWS, and az, abfs, abfss, or azure to AZURE. An unknown scheme is billable only when the matching rfgen.stores plugin lock entry declares the plan’s billing provider. This prevents a path from being costed under a different provider.

When retention_hours is supplied, each retention_months endpoint must be exactly retention_hours / Decimal(730). Mapping uses that same 730-hour rule for GB_MONTH demand. LOCAL_NONBILLABLE storage requires a file or local scheme, no provider or SKU, zero egress, and a LocalNonbillableReason; it remains in the result provenance but is not priced.

Map resource demand¶

map_admission_resource_spec(
    spec: AdmissionResourceSpecV1,
) -> tuple[tuple[ResourceDemandV1, ...], tuple[NonbillableResourceV1, ...]]

Mapping is deterministic. It emits VCPU and optional GPU hours from executor duration and count intervals, storage and egress rows for billable storage, and request and token rows for an annotation plan. Input and output token intervals are summed endpoint by endpoint. The function returns rows sorted by their canonical identities and never replaces a missing SKU, region, quota, or unit with a nearby alternative.

Verify evidence and estimate¶

estimate(
    spec: AdmissionResourceSpecV1,
    snapshot: PricingBundleV1,
    *,
    quota: QuotaSnapshotV1,
    pricing_bundle: Bundle,
    pricing_policy: AttestationPolicyV1,
    adapter: PricingVerificationAdapter,
    as_of: datetime,
    deadline_monotonic_s: float,
) -> CostEstimateV1

PricingVerificationAdapter is a narrow injected verifier. Its verify(...) method receives the Sigstore Bundle, the AttestationPolicyV1, and PricingBundleV1 as the predicate model. The application or deployment owns construction of this adapter, including trust roots, identity policy, network transport, and Sigstore configuration. rfgen does not create a default adapter or silently accept an unverified bundle.

Before mapping or pricing, estimate(...) requires a policy with predicate PRICING_V1, subject name pricing-bundle.json, and a subject digest equal to snapshot.bundle_sha256. It then requires the adapter’s verified statement to bind that exact predicate type, subject, digest, and canonical pricing payload. Any verification failure raises PricingSignatureError.

Price identity is exact on provider, SKU, region, and unit. A price is stale after 24 hours, FX is stale after seven days, and quota is stale after one hour. Pricing tiers are integrated endpoint by endpoint; foreign-currency endpoints are converted independently. Missing or stale price, FX, and quota inputs produce a typed MissingCostInputV1 and a non-COMPLETE status rather than an invented estimate. The monotonic deadline is checked throughout the operation and expiry raises AdmissionDeadlineError.

Error boundary¶

Error	Meaning
`AdmissionInputError`	The resolved configuration or provenance cannot be proved. Its code is `CONFIG_MISMATCH` or `PROVENANCE`.
`PricingSignatureError`	The DSSE policy, bundle, verified statement, or bound pricing payload is not exact.
`PricingIntegrityError`	A price, FX, or pricing-bundle canonical integrity invariant failed.
`PricingConflictError`	Price or FX evidence has duplicate or unsorted identity rows.
`QuotaIntegrityError`	Quota evidence is malformed, conflicts, is timestamp-invalid, or would require unsafe unit conversion.
`AdmissionDeadlineError`	The supplied finite monotonic deadline elapsed before an estimate could be emitted.