Jobs and production work¶
A job is durable production knowledge that work is wanted, in progress, or did not succeed. It is not a scheduler task and it is not provenance. Applications, scripts, and render services can all act as workers while keeping their own queues and process-management policies.
Every job records:
an open-world, namespaced kind such as
org.postproject:generate-proxy;the input representations;
typed parameter metadata targeting the job;
the requested output asset, representation kind, and optional media root; and
its current lifecycle state.
Unknown valid job kinds and metadata vocabularies round-trip unchanged.
Lifecycle¶
stateDiagram-v2
direction LR
[*] --> requested: request
requested --> claimed: claim
claimed --> claimed: renew, or claim after the lease expired
claimed --> requested: release
claimed --> succeeded: complete
claimed --> failed: fail
claimed --> cancelled: cancel
requested --> cancelled: cancel
succeeded --> [*]
failed --> [*]
cancelled --> [*]
An expired lease leaves the stored state claimed, but its worker can no longer renew, release, complete or fail it. Another worker can acquire a new lease.
A claim records descriptive tool and optional agent identity and an expiry. The worker owns a production-bound lease; its private credential fences earlier workers. Ordinary job facts and events do not reveal that credential.
Claim and renewal accept positive whole-microsecond durations up to 24 hours. Storage samples its own clock and checks the lease again before commit. Its durable clock high-water mark rejects guarded operations after a backward clock jump until time catches up. A forward jump can expire a lease immediately. This coordinates local workers; it is not authentication or trusted time.
Each transition is a semantic revision event. Claim tokens are deliberately absent from events: an event tells a reader to reload the job, but does not publish the capability needed to change it.
Completion and failure¶
A successful worker stages the output representation and resources, a complete
provenance activity, and the job completion in one transaction. Storage
captures the activity’s input and output snapshots during that commit. Either
the output, activity, snapshots, and succeeded state all become durable, or
none of them do.
Failure records a bounded diagnostic and no activity or representation. Activities remain facts about completed operations; pending and failed work is represented only by jobs.
Regeneration plans¶
For an existing artifact with exactly one producing activity, regeneration planning derives a new requested job from that activity’s inputs and typed parameter metadata. When the activity completed a job, the plan repeats that job’s kind and target media root; otherwise it uses the activity’s kind and no target root. The proposed output targets the existing artifact’s asset and representation kind.
Planning is read-only. It deduplicates repeated artifact IDs, does not persist the proposed jobs, does not advance the revision feed, and never runs a tool. The caller chooses whether to enqueue a proposal in an explicit transaction.
Artifact knowledge, availability, and work remain independent dimensions. A stale proxy can be online with no pending job; a requested job does not make an artifact stale.
Across public surfaces¶
The example requests work, filters by exact state and kind, and follows opaque one-item pages to completion:
static pp_error_code_t request_and_page_jobs(pp_production_t *production,
const pp_representation_id_t *input_id,
const pp_asset_id_t *output_asset_id,
pp_error_t **error) {
const char *kind = "org.example:generate-proxy";
pp_transaction_t *transaction = NULL;
pp_job_id_t job_ids[2];
pp_error_code_t status =
pp_production_begin_edit(production, NULL, NULL, &transaction, error);
for (uint32_t i = 0; status == PP_OK && i < UINT32_C(2); ++i) {
status = pp_transaction_request_job(
transaction, kind, input_id, UINT64_C(1), *output_asset_id,
PP_REPRESENTATION_PROXY, NULL, &job_ids[i], error);
}
if (status == PP_OK) {
pp_commit_receipt_t commit_receipt;
status = pp_transaction_commit(transaction, &commit_receipt, error);
}
pp_job_set_t *first_job = NULL;
pp_job_t job = {0};
if (status == PP_OK) {
status = pp_production_job(production, job_ids[0], &first_job, error);
}
if (status == PP_OK) {
status = pp_job_set_get(first_job, UINT64_C(0), &job, error);
}
if (status == PP_OK && job.state != PP_JOB_REQUESTED) {
status = PP_ERROR_INTERNAL;
}
pp_job_set_release(first_job);
char cursor_storage[2049] = {0};
const char *cursor = NULL;
uint64_t count = 0;
while (status == PP_OK) {
pp_job_set_t *page = NULL;
status = pp_production_jobs(production, PP_JOB_REQUESTED, kind,
UINT32_C(1), cursor, &page, error);
count += status == PP_OK ? pp_job_set_count(page) : 0;
const char *next =
status == PP_OK ? pp_job_set_next_cursor(page) : NULL;
if (next == NULL) {
pp_job_set_release(page);
break;
}
const int copied = snprintf(cursor_storage, sizeof cursor_storage, "%s", next);
pp_job_set_release(page);
if (copied < 0 || (size_t)copied >= sizeof cursor_storage) {
status = PP_ERROR_INTERNAL;
break;
}
cursor = cursor_storage;
}
if (status == PP_OK && count != UINT64_C(2)) {
status = PP_ERROR_INTERNAL;
}
pp_transaction_release(transaction);
return status;
}
void request_and_page_jobs(postproject::Production &production,
const postproject::RepresentationId &input_id,
const postproject::AssetId &output_asset_id) {
const postproject::JobRequest request{
"org.example:generate-proxy", {input_id}, output_asset_id,
postproject::RepresentationKind::proxy, std::nullopt};
auto transaction = production.edit().value();
const auto job_id = transaction.requestJob(request).value();
transaction.requestJob(request).value();
transaction.commit().value();
require(production.job(job_id).value().stateKind() == postproject::JobState::requested,
"requested job read back by identity");
std::optional<std::string> cursor;
std::size_t count = 0;
do {
const auto page = production.jobs(
1, cursor, postproject::JobState::requested,
std::string_view("org.example:generate-proxy")).value();
count += page.items.size();
cursor = page.next_cursor;
} while (cursor.has_value());
require(count == 2, "two requested proxy jobs");
}
def request_and_page_jobs(
production: Production,
input_id: RepresentationId,
output_asset_id: AssetId,
) -> None:
request = JobRequest(
"org.example:generate-proxy",
(input_id,),
output_asset_id,
RepresentationKind.PROXY,
)
with production.edit() as transaction:
job_id = transaction.request_job(request)
transaction.request_job(request)
transaction.commit()
assert production.job(job_id).state is JobState.REQUESTED
cursor = None
count = 0
while True:
page = production.jobs(
limit=1,
cursor=cursor,
state=JobState.REQUESTED,
kind="org.example:generate-proxy",
)
count += len(page.items)
cursor = page.next_cursor
if cursor is None:
break
assert count == 2
fn request_and_page_jobs(
production: &mut SqliteProduction,
input_id: RepresentationId,
output_asset_id: AssetId,
) -> Result<()> {
let kind = JobKind::new("org.example:generate-proxy")?;
let output = RequestedJobOutput::new(output_asset_id, RepresentationKind::Proxy, None)?;
let requested = [
Job::new(JobId::new(), kind.clone(), vec![input_id], output.clone())?,
Job::new(JobId::new(), kind.clone(), vec![input_id], output)?,
];
{
let mut transaction = production.begin_transaction()?;
for job in &requested {
transaction.request_job(job)?;
}
transaction.commit()?;
}
assert_eq!(production.job(requested[0].id())?, requested[0]);
let query = JobQuery::new(Some(JobStateKind::Requested), Some(kind));
let mut cursor = None;
let mut count = 0;
loop {
let page = production.jobs(&query, &QueryPageRequest::new(1, cursor)?)?;
count += page.items().len();
cursor = page.next_cursor().cloned();
if cursor.is_none() {
break;
}
}
assert_eq!(count, 2);
Ok(())
}
JOB_ID=$(postproject --json job request production.pproj \
org.example:generate-proxy "$ASSET_ID" proxy --input "$ORIGINAL_ID" | jq -r .id)
postproject job request production.pproj org.example:generate-proxy \
"$ASSET_ID" proxy --input "$ORIGINAL_ID"
postproject --json job show production.pproj "$JOB_ID" | jq -r .state
CURSOR=
while :; do
ARGS=(--json job list production.pproj --state requested \
--kind org.example:generate-proxy --limit 1)
[[ -n "$CURSOR" ]] && ARGS+=(--cursor "$CURSOR")
PAGE=$(postproject "${ARGS[@]}")
jq '.items[] | {id, kind, state}' <<<"$PAGE"
CURSOR=$(jq -r '.next_cursor // empty' <<<"$PAGE")
[[ -z "$CURSOR" ]] && break
done
A worker claims a job, stages the output and its activity, and completes the job so that all three become durable together:
static pp_error_code_t
complete_proxy(pp_production_t *production, pp_job_id_t job_id,
const pp_representation_id_t *input_id, const pp_asset_id_t *asset_id,
const char *output_path, pp_representation_id_t *out_proxy_id,
pp_error_t **error) {
const int64_t started = NOW + 10 * MINUTE;
const int64_t finished = NOW + 12 * MINUTE;
pp_transaction_t *transaction = NULL;
pp_job_lease_t *lease = NULL;
pp_activity_id_t activity_id = {{0}};
pp_error_code_t status =
pp_production_begin_edit(production, NULL, NULL, &transaction, error);
if (status == PP_OK) {
status = pp_transaction_claim_job_lease(transaction, job_id, "Example Transcoder",
"3.2", "https://example.com/transcoder",
NULL, NULL, NULL, NULL, 5 * MINUTE, &lease, error);
}
if (status == PP_OK) {
pp_commit_receipt_t commit_receipt;
status = pp_transaction_commit(transaction, &commit_receipt, error);
}
pp_transaction_release(transaction);
transaction = NULL;
if (status == PP_OK && !run_transcoder(output_path)) {
status = PP_ERROR_IO;
}
/* Stage the output and the activity, then complete in the same
* transaction; never commit them earlier. */
if (status == PP_OK) {
status = pp_production_begin_edit(production, NULL, NULL, &transaction, error);
}
if (status == PP_OK) {
status = pp_transaction_add_file_representation(transaction, *asset_id, PP_REPRESENTATION_PROXY, output_path, out_proxy_id, error);
}
if (status == PP_OK) {
const pp_activity_edge_t inputs[] = {{*input_id, NULL}};
const pp_activity_edge_t outputs[] = {{*out_proxy_id, NULL}};
status = pp_transaction_create_activity(transaction,
&(pp_activity_recipe_t){PROXY_JOB, inputs, 1, &started, &finished, "Example Transcoder", "3.2", "https://example.com/transcoder", NULL, NULL, NULL, NULL, NULL, 0},
outputs, 1, &activity_id, error);
}
if (status == PP_OK) {
status =
pp_transaction_complete_job_lease(transaction, lease,
*out_proxy_id, activity_id, error);
}
if (status == PP_OK) {
/* Parameters on the activity make the artifact reproducible. */
pp_object_ref_t activity;
status = pp_object_ref_from_activity(activity_id, &activity, error);
if (status == PP_OK) status = stage_proxy_parameters(transaction, &activity, error);
}
if (status == PP_OK) {
pp_commit_receipt_t commit_receipt;
status = pp_transaction_commit(transaction, &commit_receipt, error);
}
pp_job_lease_free(lease);
pp_transaction_release(transaction);
return status;
}
postproject::RepresentationId complete_proxy(postproject::Production &production,
const postproject::Job &job,
const std::string &output_path) {
auto claim = production.edit().value();
auto lease = claim.claimJobLease(job.id, encoder, std::chrono::minutes(10), worker).value();
claim.commit().value();
std::ofstream(output_path, std::ios::binary) << "encoded proxy";
// Output, provenance, and the job transition commit together or not at all.
auto transaction = production.edit().value();
const auto proxy_id = transaction.addRepresentation(
job.output_asset_id, job.output_representation_kind, output_path).value();
postproject::ActivitySpec activity{};
activity.kind = job.kind;
for (const auto &input : job.inputs) {
activity.inputs.push_back(input);
}
activity.outputs = {proxy_id};
activity.tool = encoder;
activity.agent = worker;
const auto activity_id = transaction.createActivity(activity).value();
// Parameters on the activity let the artifact be regenerated later.
transaction.addMetadataValue(
postproject::ObjectRef::activity(activity_id), transcode, "profile",
postproject::MetadataValue::plainString("editing-proxy")).value();
transaction.completeJobLease(lease, proxy_id, activity_id).value();
transaction.commit().value();
return proxy_id;
}
def run_proxy_job(production: Production, job_id: JobId, output: Path) -> None:
with production.edit() as transaction:
lease = transaction.claim_job_lease(
job_id, WORKER, timedelta(minutes=5), agent=AGENT
)
transaction.commit()
with lease:
job = job_by_id(production, job_id)
output.write_bytes(b"720p proxy essence\n") # the actual work
# Stage the output, the activity, and the completion in one transaction;
# never commit the representation or activity separately.
with production.edit() as transaction:
proxy_id = transaction.add_representation(
job.output_asset_id, job.output_representation_kind, output
)
activity_id = transaction.create_activity(
ActivitySpec(
job.kind,
inputs=tuple(item for item in job.inputs),
outputs=(proxy_id,),
tool=WORKER,
agent=AGENT,
)
)
# Copy the job parameters so the activity can be reproduced.
for parameter in production.metadata[JobRef(job_id)]:
transaction.add_metadata(
ActivityRef(activity_id), parameter.property, parameter.value
)
transaction.complete_job_lease(lease, proxy_id, activity_id)
transaction.commit()
fn complete_proxy_job(
production: &mut SqliteProduction,
job_id: JobId,
output_path: &Path,
) -> Result<(RepresentationId, ActivityId)> {
let tool = ToolIdentity::new("Example Worker", Some("1.0".to_owned()), None)?;
let job = production.job(job_id)?;
let claim = {
let mut transaction = production.begin_transaction()?;
let claim = transaction.claim_job_lease(job_id, &tool, None, Duration::from_secs(60))?;
transaction.commit()?;
claim
};
// ... the worker writes the proxy file to output_path here ...
let output = prepare_representation(
job.requested_output().asset_id(),
RepresentationKind::Proxy,
output_path,
)?;
let activity = Activity::new(
ActivityId::new(),
ActivityKind::new(GENERATE_PROXY_JOB_KIND)?,
job.inputs()
.iter()
.map(|input| ActivityInput::new(*input, None))
.collect(),
vec![ActivityOutput::new(output.representation().id(), None)],
)?
.with_tool(tool);
let parameters = production.metadata(ObjectRef::Job(job_id))?;
let mut transaction = production.begin_transaction()?;
// Output, activity, and the succeeded state commit together or not at all.
transaction.complete_job_lease(&claim, &output, &activity)?;
// Keep the parameters on the activity so the artifact can be regenerated.
for parameter in ¶meters {
transaction.add_metadata_value(
ObjectRef::Activity(activity.id()),
parameter.property(),
parameter.value(),
)?;
}
transaction.commit()?;
Ok((output.representation().id(), activity.id()))
}
postproject --json job claim jobs.pproj "$JOB_ID" \
--tool-name "Proxy Worker" --tool-version 1.0 \
--lease 5m --lease-token-file completion.token
printf 'proxy media' > proxies/A001_proxy.mov
# Adds the output representation and its producing activity, and marks the
# job succeeded, in one transaction.
COMPLETED=$(postproject --json job complete jobs.pproj "$JOB_ID" \
proxies/A001_proxy.mov --lease-token-file completion.token)
jq '{state, completion_representation_id, completion_activity_id, commit_receipt}' <<<"$COMPLETED"
See bounded queries and cursors for continuation rules.
See artifact knowledge and reproducibility for the derived knowledge state and jobs and workers for the worker protocol. The reference local executor is one optional worker; jobs do not depend on it and are never run implicitly.