Report execution at scale

Reporting stores the grant id on the definition beside the scheduler job id. That grant is the standing consent for the schedule. When a trigger admits a run, reporting copies the grant id onto the report instance, so the run acts under the grant that was live at admission. A later schedule, unschedule or party change therefore cannot change the authority of a run that is already in flight.

• Schedule. A person asks reporting to schedule a definition. Reporting creates the grant on behalf of the person, with their token, and then puts the scheduler job.
• Manual run. A person triggers a run. Reporting creates a grant for that one run on behalf of that person, so the run acts with the authority of the person who triggered it, and not with the schedule's.
• Change of party. A change to the definition's party revokes the grant and creates a new one. The person who makes the change gives the new consent.
• Unschedule or delete. Reporting deletes the job and revokes the grant.
• Lapse. A step reports grant_lapsed. Reporting fails the run and moves the definition to suspended. A tenant administrator, or another person of the party, schedules the definition again, which is a new consent.

The role holds only what a run needs. Each owner seeds its own permission codes, and IAM seeds the role.

The role has no write on reference data, trades or market data. A run that tries one is refused by the owner's permission check.

At the design targets below, 1,000 runs in flight for thirty minutes each make about four exchanges a second, and 2,000 runs that start in one minute make about 100 exchanges in that minute. Run grants and run tokens states why one IAM replica serves this.

A person of the party creates the definition in reporting, with their own token. The definition holds the report type, the scope, the concurrency policy and the cron expression. The engine's configuration lives with its owners. The person imports it with ore.v1.ops.import_run_configuration. ORE runs that import as a saga over the owners: reporting saves the run document, and refdata and analytics save the configuration documents. Reporting binds each document to the definition. Every write uses the person's token, so every row belongs to the party.

No grant exists yet. A definition without a schedule cannot run, except by a manual trigger.

The person asks reporting to schedule the definition. Reporting performs three actions, in this order:

1. It creates the grant through IAM, on behalf of the person.
2. It puts the scheduler job. The job name is report_definition.<id>. The job holds the tenant and the party, so the firings belong to the tenant, and a tenant administrator can see them.
3. It records the grant id and the job id on the definition.

Each action is idempotent. A repeated grant create for the same definition and grantor returns the active grant. The job name is its identity, so a repeated put changes nothing. A failure after step 1 leaves a grant that no job uses. The next schedule request adopts it.

Reconciliation is as described on the definition page, with two additions. A definition with a job and no active grant is suspended. A job whose definition is gone is deleted, and its grant is revoked.

Source: report_definition_schedule_states.puml. Regenerate with plantuml -tpng doc/knowledge/architecture/report_definition_schedule_states.puml.

The scheduler loop fires the job at its minute. It writes the firing record and then sends the trigger to reporting as an authenticated request with its own identity. The trigger carries the definition, the tenant and the firing id. The scheduler's part ends with the reply.

The trigger needs no tenant identity. Reporting checks that the caller holds the trigger permission, then acts on the definition as itself. That is reporting's own work on its own table. The run's tenant identity starts with the first step.

Reporting decides the state the instance is born in. Two checks apply, in this order:

1. The definition's concurrency policy. It applies when another run of the same definition is in flight: skip, fail or queue, as the definition page describes.
2. The quotas. A tenant has a limit on its running instances, and a party has a limit inside its tenant. A run over either limit is born queued.

The quotas are configuration in reporting, per tenant, with a platform default. A firing never fails because of a quota. It queues.

Reporting promotes queued runs when a run of the tenant reaches a terminal state. Promotion takes the oldest queued run of the tenant whose party is under its limit, moves it to pending, and starts its workflow. Promotion uses a row lock on the oldest candidate, so two replicas cannot promote the same run. A sweep every minute promotes any run that an event missed.

Two replicas can admit runs of one tenant at the same time. Admission therefore takes a transaction-scoped advisory lock on the tenant before it counts. The lock serialises admission inside one tenant and never across tenants.

Source: report_admission_interaction.puml. Regenerate with plantuml -tpng doc/knowledge/architecture/report_admission_interaction.puml.

Source: report_instance_admission_states.puml. Regenerate with plantuml -tpng doc/knowledge/architecture/report_instance_admission_states.puml.

The trigger is idempotent on the firing id. A repeated trigger finds the instance for that firing and returns it.

Reporting creates the instance and publishes workflow.v1.ops.start_workflow. The workflow instance id is created by reporting and recorded on the instance. A repeated start therefore opens the same run. The instance records the grant id it was admitted under, and the workflow request carries that same id, so a step service exchanges it for a run token without reading reporting. The request carries:

The as-of instant fixes the shape of the run. The definition page requires that an edit during a run does not change that run. The steps read every bitemporal document as of this instant, so an edit after the firing does not reach the run. The start message therefore carries a reference and an instant, and not a copy of the documents.

The chain for a run that executes both ends has four steps:

Pre-processing is one step, as the pre-processing page requires. The substituted chains replace step 1 or step 3 as Report execution substitution describes. Every step compensates with reporting.v1.ops.fail_report.

ores.ore owns this step. It owns the mapping to the engine's documents, so the archive has one author.

1. Identity. It obtains the run token.
2. The run document. It calls reporting.v1.run_documents.get for the definition, as of the instant. The answer names the configuration documents the run binds.
3. The configuration. It calls the owners' gets in parallel: refdata.v1.curve_configuration_documents.get, refdata.v1.conventions_documents.get, analytics.v1.pricing_engines_documents.get and analytics.v1.todays_market_documents.get, each with the party and the instant.
4. The digest. It computes the digest of the tenant, the party, the version of each document, the scope, the as-of instant, the market data instant and the engine version. The tenant and the party are in the digest, so two tenants never share an archive.
5. The lookup. It asks storage for ore/inputs/<digest>.tar.gz. A complete object ends the step: the run reuses it.
6. The data. It calls the trading scope read and the market data resolution, as Report pre-processing specifies. Each owner writes its batch to storage under the run's key with the run token and answers with the key. Bulk data does not travel in a message.
7. Map and package. It streams the batches through the mappers into the archive and writes it under the digest key. A write goes to a temporary key and is then renamed, so a reader never sees a partial archive.
8. Answer. It completes the step with the key, the digest, and whether it generated the archive or found it.

The step converges on a repeat. A repeat finds the archive under the digest, or it generates it again under the same key.

ores.compute owns this step. It obtains the run token and creates the batch in the tenant, with the report instance id as the batch's external reference. A repeat finds the batch by that reference and adopts it, so a repeat never runs the engine twice. The workunits name the input key, the package of the tenant's engine version, and the output key.

The step does not complete at submission. Compute records a link between the batch and the step. When the batch closes, compute completes the step with the batch id and the output keys. The step's deadline is the grid budget, which covers the queue wait and the engine time.

Compute dispatches with its own identity. Dispatch is compute's own work on its own table, and the fair share below decides its order.

A worker node takes an assignment from the queue of its platform. The worker holds no run token. An engine run can last longer than any token, and a node serves many tenants. The node uses its own identity. That identity reads only compute/packages/ and ore/inputs/ and writes only compute/output/. Each assignment names the exact keys. The node runs the engine and writes the output archive, then submits the result on compute.v1.ops.submit_result. draws this path beside a step service that keeps exchanging. Compute job lifecycle describes the node's work.

Reporting obtains the run token and reads the output archive from storage. It uses the parser that ores.ore supplies as a library, and it bulk loads the results into the result store as Report post-processing describes. Each row carries the instance id, the tenant and the party. A repeat replaces the rows of the instance and does not add to them.

Reporting moves the instance to completed and records the output key, the row counts and which phases were substituted. It then promotes the next queued run of the tenant. The engine publishes the instance status event. Every step service drops the run's cache entries when it receives that event.

These numbers are targets for capacity tests. They are not measurements.

A busy minute costs little at the front, because the trigger and the start do no heavy work. The heavy work waits in the step queues and in the grid, and those queues are fair.

• Admission, in reporting. The quotas cap the running runs of each tenant and each party. Promotion is first in, first out inside a tenant. A tenant that schedules 500 reports at the same minute runs 20 at a time. The other tenants are not blocked behind it.
• Step concurrency, in each step service. Each replica has a fixed pool of workers for heavy steps. It also limits the steps of one tenant that run at the same time on the replica. A step over the tenant limit waits in a local queue, and the replica serves its local queues round robin across tenants. The wait counts against the step's deadline, so the limits must keep the wait well inside the budget.
• Dispatch, in compute. Compute holds the workunits and releases them to the platform queue by weighted fair share across tenants. A tenant has a weight and a cap on its assignments in flight. A tenant with dedicated nodes has a pool of its own, and its work does not enter the shared pool.

The engine time of a run is the largest cost, so the grid's fair share decides most of the waiting. Admission keeps the grid queue short, so a run waits in queued, where a person can see it, and not in a hidden queue.

Each service has its own replicas, pools and limits. A slow owner delays only the step that calls it. The step's deadline bounds that delay, and the run fails with a message that names the service, as Failure, compensation and retry describes. A slow tenant cannot take all the workers of a replica, because of the tenant limit in the step pool.

None of these caches holds data across tenants. The input digest includes the tenant and the party, and the token cache key includes the grant.