Job Orchestration (Inngest)
Relevant source files
The following files were used as context for generating this wiki page:
Job orchestration in ZINC-Fusion-V15 is handled by Inngest, a serverless event-driven system running on Vercel. It manages the scheduling, execution, and monitoring of approximately 88 different functions responsible for data ingestion, market intelligence updates, and periodic maintenance tasks. The system is designed to provide institutional-grade quantitative forecasting by ensuring a reliable flow of data from over 847 global sources into the Prisma Postgres data warehouse.
Sources: README.md:162-165, cubic.yaml:23-24
Core Configuration and Constraints
The Inngest implementation is governed by strict isolation and concurrency rules to prevent cross-project contamination and database exhaustion. The system utilizes a centralized client identified as fusion-jobs.
Infrastructure Guardrails
A critical constraint is the DB_CONCURRENCY limit, which is capped at 5 with an "env" scope. This limit prevents the 88 concurrent functions from exhausting the 50-connection Postgres pool. Furthermore, all side effects such as database queries or API fetches must be wrapped in step.run() to avoid duplicate executions during Inngest's durable execution replay process.
| Parameter | Value | Description |
|---|---|---|
| Client ID | fusion-jobs | Must be specific to V15 to avoid Rabid Raccoon project contamination. |
| DB_CONCURRENCY | 5 (scope: "env") | The maximum number of functions allowed to query Postgres simultaneously. |
| Step Isolation | Required | All fetch(), db queries, and stateful logs must reside inside step.run(). |
| Registration | Required | New functions must be exported from functions.ts and added to route.ts. |
Sources: cubic.yaml:45-53, cubic.yaml:58-62, cubic.yaml:67-76
Ingestion Workflow and Pipeline
The orchestration of ingestion follow a standardized "Ingestion Contract" to ensure data integrity and observability. This pipeline typically involves four to five distinct steps managed by Inngest.
Standardized Ingestion Pipeline
The following diagram illustrates the standard data flow for an Inngest function in this repository:
The diagram shows the standard flow including quarantine and deduplication logic required by the system's "Ingestion Contract".
Sources: frontend/src/inngest/federal-register.ts:167-240, frontend/src/inngest/fred-daily.ts:390-445
Ingestion Contract Requirements
Every function writing to the database must adhere to the following rules:
- Logging: Must create a record in
ops.ingest_runtracking attempted, inserted, skipped, and quarantined rows. - Deduplication: Must use an
ON CONFLICT DO UPDATEpattern with arow_hash(SHA256) for change detection. - Quarantine: Invalid records (NaN prices, missing dates) must be moved to
ops.quarantined_recordinstead of being silently dropped. - Append-Only: Landing tables (e.g.,
mkt.*,econ.*) must be append-only;DELETEoperations are strictly forbidden.
Sources: cubic.yaml:83-93, cubic.yaml:98-103, cubic.yaml:108-115
Specialized Ingestors
The system orchestrates specialized ingestors for various domain-specific data sources, each with its own scheduling and logic.
Economic and Policy Ingestors
Data from sources like the Federal Reserve (FRED) and the Federal Register are partitioned into segments to manage load and provide granular tracking.
| Ingestor | Frequency | Specialist Tags | Table Routing |
|---|---|---|---|
| FRED Fed | Every 8 hours | fed, fx | econ.rates_1d |
| FRED Energy | Every 8 hours | energy, biofuel | econ.commodities_1d |
| Federal Register | Daily (03:15 UTC) | tariff, trump_effect, biofuel | alt.legislation_1d |
| Farmdoc RINs | Daily (01:30 UTC) | biofuel, energy | alt.policy_news_event |
Sources: frontend/src/inngest/fred-daily.ts:257-275, frontend/src/inngest/federal-register.ts:159-165, frontend/src/inngest/farmdoc-rins.ts:28-30
Logic and Specialist Classification
The federal-register.ts ingestor demonstrates the use of a hybrid classification approach where raw text from government documents is parsed using regex patterns to assign specialist_tags. This ensures that downstream ML models (Specialists) receive only relevant filtered signals.
// Example Tagging Rule for Federal Register
{ pattern: /section[\s_-]?301/i, tags: ["tariff"] },
{ pattern: /renewable[\s_-]?fuel[\s_-]?standard|rfs/i, tags: ["biofuel"] }
Sources: frontend/src/inngest/federal-register.ts:50-100
Monitoring and Data Integrity
The orchestration system includes mechanisms for detecting "stale" runs and handling failures.
Execution Safety
Inngest functions are required to refresh database connection pools across step boundaries. This prevents "stale connections" that occur when Inngest pauses execution for several minutes between durable steps. Additionally, all external API calls must include explicit timeouts to prevent a single hanging request from blocking a DB_CONCURRENCY slot.
Sources: cubic.yaml:168-174, cubic.yaml:209-215
Error Handling Sequence
The following sequence diagram represents the interaction between Inngest, the External API, and the internal operations tables during a failure:
The job orchestration system via Inngest is the backbone of the ZINC-Fusion-V15 data pipeline. By enforcing strict concurrency limits, standardized ingestion contracts, and isolated step execution, it ensures that the 19-model ML stack is fed with high-integrity, deduplicated data across 12 different database schemas.