flowctl: re-tool preview on the runtime-next stack#3117
Open
jgraettinger wants to merge 3 commits into
Open
Conversation
jgraettinger
changed the title
preview on the runtime-next stack
jgraettinger
force-pushed
the
johnny/preview
branch
4 times, most recently
from
036c2b9 to
f6adb49
Compare
…tories Factor runtime-next's host-specific behavior — how a task publishes, shuffles, and logs — out of the crate into three uniform, monomorphized host seams, each a factory the leader and shard `Service`s are generic over: - publish: replace the `Publisher` enum — a `Real` variant plus a `Preview` variant that baked NDJSON-to-stdout output into the crate — with a `PublisherFactory` / `JournalPublisher(Factory)` seam. The publisher now carries only document output; its former observation calls move to the Logger seam. - leader: factor the leader's inline shuffle-`Session` opening and checkpoint sourcing into a new `leader/shuffle.rs`, providing `ShuffleSession` / `ShuffleSessionFactory` (static RPITIT, not `dyn`) plus a `ShuffleServiceFactory`. - log: new `Logger` / `LoggerFactory` seam — the task's out-of-band log and event stream. A `Logger` sinks both the connector's own logs (`log`) and structured runtime `LogEvent`s (`event`): a `#[non_exhaustive]` enum (persist / applied / inferred-schema / container lifecycle) whose variants carry borrowed, verbatim payloads so a host may intercept them structurally. Its canonical `LogEvent::to_log` flattens each event into the ops-log line it replaced, and `event` defaults to that flattening — so a host overrides only to intercept, delegating the rest. Production shards install `FnLoggerFactory` (sinking to the task-log writer); leaders and tests install `TracingLoggerFactory`. Remove the now-unnecessary `Task.preview` flag from the runtime proto (reserving field 2): how a task publishes and logs is a host concern decided by the installed factories, not a protocol flag. `max_transactions` stays as the transaction bound, re-documented as a general preview/harness control rather than tied to the removed flag. Make the `shard::rocksdb` module public — exposing `RocksDB` (with its `open`, `scan`, and now-public `put_connector_state_base`) and re-exporting its `DecodeError` — so a host harness can seed shard zero's RocksDB by path and read back its final reduced state itself, and re-expose `shuffle::log::BlockMeta` for harness segment writers.
Drive `flowctl raw preview-next` through runtime-next's three factory seams by installing preview-specific implementations of each: - publish: an output-capturing `PreviewPublisherFactory` writes captured / derived documents to stdout as NDJSON. - log: a `PreviewLoggerFactory` sinks the task's log stream and powers `--output-state` / `--output-apply` (the legacy `connectorState` / `applied` lines) by intercepting runtime-next's `LogEvent::Persist` / `LogEvent::Applied` structurally — decoding state deltas from its tab-framed patch wire form — and flattening every other event through `LogEvent::to_log` into the same log handler as the connector's own logs. The `--output-state` run-end final reduced state is emitted by re-opening shard zero's RocksDB (`read_preview_state`) once the session loop has released it. - shuffle: a `PreviewShuffleFactory` enum selects per run between a live in-process journal-reading shuffle Session and a `--fixture` replay (`FixtureOpener`, a `ShuffleSessionFactory` that writes log segments and feeds synthetic checkpoint Frontiers). `--initial-state` seeds shard zero's RocksDB by path, and `--output-state`'s run-end read reopens it — both managed by preview itself through the now-public `runtime_next::shard::rocksdb` handle (the local `seed_preview_state` / `read_preview_state` helpers over `RocksDB::open`, `put_connector_state_base`, and `scan`). Each driver also encodes its task spec once and threads the shared bytes to its shards rather than re-encoding per shard. `--delay`, eager/streaming fixtures, and `--sessions` planning are unchanged.
`flowctl preview` now runs on the runtime-next + shuffle stack that was
developed on this branch as `flowctl raw preview-next`, which the prior
commits verified to be behaviorally compatible with the legacy preview.
Move the `raw/preview_next/` module into `preview/`, delete the old
V1-runtime `preview/{mod,journal_reader}.rs`, and drop the `raw
preview-next` command wiring. The CLI flag surface, stdout/stderr line
shapes, and exit-code contract that the estuary/connectors CI harnesses
depend on are preserved; `--shards` and `--debug-port` are additive.
The old preview was the sole consumer of `runtime::harness` (and its
private `exchange` combinator), so remove that now-dead module too.
Update the soak README to the new command name, and delete the
runtime-v2 preview harness guide: a closed, point-in-time document
describing `raw preview-next` as it stood during bring-up.
jgraettinger
force-pushed
the
johnny/preview
branch
from
f6adb49 to
53cacd6
Compare
| schema, | ||
| .. | ||
| } => { | ||
| fields.push(("collection", json_field(collection_name))); |
Member
There was a problem hiding this comment.
Wanting to confirm this field name...should it be collection_name? Since the L1 inferred schemas rollup is looking for that:
| // The patch payload is valid JSON (a tab-delimited JSON array; | ||
| // see `crate::patches`), so it embeds verbatim. | ||
| fields.push(("patches", persist.connector_patches_json.clone())); | ||
| (ops::LogLevel::Debug, "persisted connector-state delta") |
Member
There was a problem hiding this comment.
Will these get filtered out for info and above level tasks, since they don't go through Tokio's tracing subscriber?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What & why
flowctl previewnow runs on the runtime-next + shuffle stack instead of thelegacy V1
runtime::harness. It was developed on this branch asflowctl raw preview-next, then promoted to be thepreview; the V1 preview harness — thelast consumer of
runtime::harness— is deleted along with the now-deadruntime/src/harness/*and its privateexchangecombinator.The user-facing contract is unchanged: the CLI flags, stdout/stderr line shapes,
and exit codes that
estuary/connectorsCI depends on.--shardsand--debug-portare new, additive flags.To get there, runtime-next's host-specific behavior — how a task publishes
documents, reads its sources, and emits logs — was factored into three factory
seams. Production installs journal-backed factories;
previewinstallsoutput-capturing / fixture-replay factories from
flowctl. runtime-next is nowpreview-agnostic, so the lone
Task.previewflag — the "write documents tooutput instead of collections" wire control — is removed from the runtime proto
(field 2 reserved); how a task publishes and logs is now a host concern of the
installed factories, not a wire flag. (
max_transactionsstays as thetransaction bound, re-documented as a general preview/harness control.)
The three host seams (
runtime-next)Each seam is a factory the leader and shard
Services are generic over (staticRPITIT, not
dyn):PublisherFactory/JournalPublisherseam replaces thePublisherenum (aRealvariant plus aPreviewvariant that bakedNDJSON-to-stdout output into the crate); the publisher now carries only
document output, with its former observation calls moved to the logger seam.
leader/shuffle.rsprovidesShuffleSession(Factory)ShuffleServiceFactory, factoring out the leader's previously-inlineshuffle-
Sessionopening and checkpoint sourcing.Logger/LoggerFactoryseam sinking both connector logs(
log) and structuredLogEvents (event) — a#[non_exhaustive]enum(persist / applied / inferred-schema / container lifecycle) carrying borrowed,
verbatim payloads so a host can intercept them structurally.
LogEvent::to_logflattens each event into the ops-log line it replaced, and
eventdefaults tothat flattening, so a host overrides only the events it cares about. Production
shards install
FnLoggerFactory; leaders and tests installTracingLoggerFactory.New harness helpers:
seed_initial_connector_state/read_final_connector_stateseed and read shard zero's RocksDB by path;
shuffle::log::BlockMetaisre-exposed for segment writers.
Preview on the seams (
flowctl)PreviewPublisherFactorywrites captured/derived documents tostdout as NDJSON.
PreviewLoggerFactorysinks the task log and powers--output-state/
--output-apply(the legacyconnectorState/appliedlines) byintercepting
LogEvent::Persist/LogEvent::Appliedstructurally andflattening everything else through
LogEvent::to_log. The run-end final reducedstate comes from re-opening shard zero's RocksDB once the session loop releases
it.
PreviewShuffleFactoryselects per run between a live in-processjournal-reading
Sessionand a--fixturereplay (FixtureOpener).--initial-stateseeds shard zero's RocksDB viaseed_initial_connector_state.--delay, fixtures, and--sessionsplanning are unchanged.Tests / soak
plans/runtime-v2/preview-harness.md, a closed bring-up doc forraw preview-next.Notes for reviewers
this branch; the preview work does not require them, so the two can land in
either order.
the promotion commit is mostly a module move (
raw/preview_next/→preview/)plus deletion of the V1 harness.