✨ Launch the ingestion feature - #208
Conversation
Interval-based batch identity, validated journal history, enforced execution deadline, abandonable readers, load permit per attempt, serialized journal appends, stateless executor for concurrent work.
One write unit per declared source interval, poisoned append chain after an unconfirmed fact, no progress commits from abandoned work, and envelope-change validation in the journal projection.
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 69324c84af
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // that changed between attempts. | ||
| let sinceBoundary = 0 | ||
| for (let batch = await queue.pop(signal); batch !== undefined; batch = await queue.pop(signal)) { | ||
| const discriminator = batch.intervalIds ? `interval:${canonicalJson(batch.intervalIds)}` : `content:${canonicalJson(batch.rows)}` |
There was a problem hiding this comment.
When a source yields rows containing JavaScript bigint values—for example for UInt64/Int64 columns—and does not declare an interval id, this call passes the rows to canonicalJson, whose JSON.stringify throws on bigint. The stream therefore fails before invoking its loader even though Row permits these values; serialize bigint deterministically or validate/narrow the supported row values.
Useful? React with 👍 / 👎.
| signal: AbortSignal, | ||
| classifier: ErrorClassifier | undefined | ||
| ): FailureClass { | ||
| if (signal.aborted || isAbortError(cause)) return { kind: 'cancelled' } |
There was a problem hiding this comment.
When a provider operation throws an AbortError from its own request-local timeout while the execution signal remains active, this unconditional check classifies it as executor cancellation before the provider classifier can handle it. executeStream then reports budget_exhausted and skips retries even though the execution budget was not exhausted; only treat the error as execution cancellation when the supplied signal is actually aborted.
Useful? React with 👍 / 👎.
| await queue.push(pending, signal) | ||
| pending = emptyBatch() | ||
| } | ||
| if (stream.budget?.maxChunks !== undefined && progress.chunks >= stream.budget.maxChunks) budgetExhausted = true |
There was a problem hiding this comment.
If loading a chunk fails after progress.chunks reaches maxChunks, the outer retry recreates the reader with a fresh budgetExhausted = false, and this limit is checked only after another chunk has already been pulled. Each reader retry can therefore fetch an additional chunk beyond the documented execution-wide maximum; check the accumulated count before calling iterator.next() or stop recreating the reader once the limit is reached.
Useful? React with 👍 / 👎.
KeKs0r
changed the title
2 participants
Add scheduled API-to-ClickHouse ingestion as the optional TypeScript
@chkit/plugin-ingestpackage. Readers yield destination-shaped rows, writes land before checkpoints advance, and failed runs resume from journaled progress with stable deduplication tokens. A later successful full sync starts a distinct batch identity cycle, so an A → B → A source update is not suppressed as a retry.Changes
defineStreamand exporteddefinePipelinedefinitions, discovered through an opt-in projectentrymodule. Only exported pipelines participate; there is no process-wide registry.ingest run,list, andstatus, exact AND tag selection, isolated backfill namespaces, timestamp-window and provider-cursor strategies, and full syncs.rawTable/rawRowsfor native JSON landing tables, runtime ingestion metadata, and optional per-insert ClickHouse settings.Compatibility
Existing schema-glob configs and CLI runtime behavior require no migration.
entryis optional and mutually exclusive with schema globs; resolvedconfig.schemaremains a string array. Existing insert calls do not need the new optional settings field.TypeScript source caveat:
ChxUserConfig.schemais now optional. External code reading that field from a value annotated asChxUserConfigmay need to normalize withresolveConfig(config).schemaor handleundefined. This is not a blanket source-compatibility guarantee for consumers of the raw config type.Ingestion requires a direct ClickHouse connection. Host-provided executors are rejected before writes because their contract does not guarantee JSON encoding and deduplication settings. This restriction applies to the new ingestion commands.
The new ingestion package and
entryoption are TypeScript-only; Python behavior is unchanged. The ingestion journal uses its existing table shape and checkpoint envelope; successful-cycle identity is projected from existing completion events.The new ingestion retry callbacks use p-retry's native context plus classified
FetchFailureerrors. The unreleased customretryDelaycontext field andExecutionEnv.sleep/randomtest hooks are removed; p-retry owns timers and jitter. Existing released feature APIs are unchanged by this simplification.Validation
git diff --checkpassed.