feat(stovepipe): wire SourceControl + stores into ingest, publish to process#279
Merged
Merged
Conversation
behinddwalls
force-pushed
the
preetam/stovepipe-ingest-integration
branch
from
3bf1077 to
f410a35
Compare
roychying
approved these changes
Jun 26, 2026
behinddwalls
added a commit
that referenced
this pull request
Jun 26, 2026
…roller (#276) ## Summary ### Why? The Stovepipe workflow RFC (PR #275, this PR's base) describes a pipeline whose entry point is **ingest**: an external poller reports that a queue (a named repo+ref) has a new commit, and Stovepipe mints a request to validate it. This PR lays the first foundation stones — the domain model and the entry RPC — as a deliberately thin, log-only slice. ### What? Three stacked commits: - **`Request` entity** (`stovepipe/entity`) — one validation of a queue at a particular commit: ID namespaced by the queue (`"request/<queue>/<counter>"`), queue name, opaque VCS-agnostic commit `URI` (empty until SourceControl resolution lands), `RequestState` (initial state `accepted`), and `Version` for optimistic locking; with `ToBytes`/`FromBytes` and a lightweight `RequestID`. - **`Ingest` RPC** (`api/stovepipe`) — `Ingest(IngestRequest{queue}) returns (IngestResponse{id})` added to the single-service Stovepipe proto; only the queue name is on the wire (commit-URI resolution via SourceControl is a follow-up). Includes regenerated protopb stubs. - **Thin `IngestController`** (`stovepipe/controller`) — validates the queue, mints the ID via the counter extension, builds + logs the `Request` (state `accepted`), returns the ID. No storage, publish, or SourceControl yet. Wired into the example server behind a minimal in-process counter. Greenness is intentionally **not** modeled here: the RFC treats it as a continuous degree (e.g. a fraction of projects broken), which an enum cannot represent, and nothing in this slice consumes it. It will be introduced with the `record`/project-analysis stage. Other explicit follow-ups: SourceControl extension + commit-URI resolution; storage + persistence; publishing onto the `process` topic; the stateful Queue entity. ## Test Plan - ✅ `bazel test //stovepipe/...` — `//stovepipe/entity` and `//stovepipe/controller` tests pass (serialization round-trips; ingest happy path / empty-queue user error / counter-error classification). - ✅ `make proto` regenerates only the stovepipe stubs; `make lint`, `make check-gazelle`, `make check-tidy` clean. - ✅ `bazel build //example/stovepipe/...` builds the wired server. ## Issues ## Stack 1. @ #276 1. #277 1. #278 1. #279
behinddwalls
force-pushed
the
preetam/stovepipe-sourcecontrol
branch
from
cc75a0b to
7b5f323
Compare
behinddwalls
force-pushed
the
preetam/stovepipe-ingest-integration
branch
from
f410a35 to
d17beeb
Compare
behinddwalls
added a commit
that referenced
this pull request
Jun 26, 2026
…Store (MySQL) (#277) ## Summary ### Why? The ingest controller (PR #276) mints a `Request` but is log-only. For the pipeline to do real work it needs to persist requests and look them up two ways, both flowing from the workflow RFC: by **request ID** (every downstream stage reloads the entity), and by **(queue, commit URI)** to find whether a commit is already being validated — the RFC's `(Queue, head URI)` dedup key. ### What? A new `stovepipe/extension/storage` extension plus its first backend (MySQL), mirroring the SubmitQueue storage conventions: a factory `Storage` interface, `ErrNotFound`/`ErrAlreadyExists`/`ErrVersionMismatch` sentinels, metrics-wrapped MySQL ops, and optimistic-locking CAS with version arithmetic owned by the caller. Two stores, **one per table**: - **`RequestStore`** (`request` table) — `Create` (ErrAlreadyExists on dup ID); `Get` by ID (ErrNotFound); `Update`, a pure conditional write taking the whole `Request` plus `oldVersion`/`newVersion`, persisting the mutable fields (uri, state) only if the stored version matches (else ErrVersionMismatch). - **`RequestURIStore`** (`request_uri` table) — the reverse index from a validated commit to its request, keyed by `(queue, uri)`: `Create` (ErrAlreadyExists on a duplicate `(queue, uri)` — the dedup signal) and `GetIDByURI` (ErrNotFound). It's a separate store because it's a separate table; the two are written independently (no cross-table transaction) so the contract stays satisfiable by key/value backends. Ships the MySQL impl, schema (`request.sql`, `request_uri.sql`), generated mocks, and a docker-compose integration test. Follow-ups: wire `storage` into `ingest` (dedup via `GetIDByURI`, then `Create`); an in-memory backend + shared contract suite. ## Test Plan - ✅ `bazel test //test/integration/stovepipe/extension/storage/mysql:mysql_test` — real MySQL via docker-compose: create/get/update-CAS (success, stale-version mismatch, missing-row mismatch), not-found, duplicate-ID, and the URI mapping (round-trip, not-found, duplicate dedup, per-queue isolation). - ✅ `bazel build //stovepipe/...`; `make check-gazelle`, `make check-tidy`, `make lint` clean. Mocks regenerated and idempotent. ## Issues ## Stack 1. #276 1. @ #277 1. #278 1. #279
behinddwalls
force-pushed
the
preetam/stovepipe-sourcecontrol
branch
from
7b5f323 to
7ce0fd1
Compare
behinddwalls
force-pushed
the
preetam/stovepipe-ingest-integration
branch
from
d17beeb to
6695ab7
Compare
behinddwalls
added a commit
that referenced
this pull request
Jun 26, 2026
## Summary ### Why? The `Request.URI` field is "empty until SourceControl resolution is wired in" — Stovepipe has no way to resolve a queue's head commit, compare two commits for history rewrites, or enumerate a ref's recent commits. The workflow RFC already names SourceControl as the sole owner of URI semantics with exactly these three responsibilities. This adds that contract so downstream stages (`ingest`, `process`) and a future status view can be built against it. ### What? New `stovepipe/extension/sourcecontrol` package holding the contract only (interface + Config + Factory interface + sentinel error), per the extension-design rules — no factory impl or routing. The `SourceControl` interface is bound to a single queue by its Factory, so methods take no queue argument: - `Latest(ctx)` — latest commit URI on the queue's ref (VCS-agnostic name, not "Head"). - `IsAncestor(ctx, ancestor, descendant)` — ancestry check; `false` is the history-rewrite signal that drives a full build. - `History(ctx, cursor, limit)` — cursor-paginated, newest-first page of commit URIs; bounded so a remote backend stays cheap. URIs and the pagination cursor are opaque tokens interpreted only by the implementation. Pagination uses a new shared generic `platform/base/page.Page[T any]` (`Items` + opaque `NextCursor`) rather than a one-off struct — the repo's first generic, reusable for any future paginated read across domains. Method-result data types live outside the extension package, matching existing precedent (`entity.BuildStatus`, `entity.Conflict`). Ships with an in-memory `fake` backend (seeded with a queue's newest-first history) and a generated gomock for use in controller and pipeline tests. ## Test Plan ✅ `bazel test //stovepipe/extension/sourcecontrol/... //platform/base/page/...` — fake unit tests (Latest / IsAncestor / History pagination + not-found cases) pass. ✅ `make build`, `make gazelle`, `make fmt` clean. ## Issues ## Stack 1. #276 1. #277 1. @ #278 1. #279
…process ## Summary ### Why? Ingest was a thin stub: it minted a request id but never resolved the commit URI, persisted anything, or moved the request onto the pipeline, so `Request.URI` stayed empty and nothing consumed the work. This makes ingest the real pipeline entry and adds the first internal queue contract so the pipeline can hand work to the next stage. ### What? Ingest now resolves the queue's head URI via the SourceControl extension, dedups on the (queue, URI) pair, persists the Request and its URI mapping via storage, and publishes the request id to a new process stage over the messaging queue. Ingestion is idempotent: a re-reported head resolves to the already-minted request and nothing is published again. The URI mapping is claimed before the request row is written, so a lost race leaves no orphan row. Adds the first internal proto message-queue contract under `stovepipe/core/messagequeue` (proto3 + protojson, mirroring `api/runway/messagequeue`): a `ProcessRequest` payload carrying the id, the `TopicKeyProcess` constant, and the protojson glue, wired into the proto codegen (`tool/proto`, `PROTO_PACKAGES`). Per CLAUDE.md, internal contracts live under the domain's `core/`, not `api/`, and the contract package owns both the payload and its topic keys. Adds a minimal `process` consumer (`stovepipe/controller/process`) that reloads the Request by id and logs it; a not-yet-visible request is retryable so redelivery converges. The build-strategy/ancestry logic the RFC assigns to `process` is deferred. Wires the example server (`example/stovepipe/server`) into a MySQL storage + MySQL queue + fake SourceControl stack with the process consumer running, plus docker-compose (two databases) and a schema-init make target. ## Test Plan ✅ `bazel test //stovepipe/...` — contract round-trip + topic-key binding, ingest (happy/dedup/race/unknown-queue/infra-error paths), and process consumer unit tests. ✅ `bazel test //test/integration/stovepipe:stovepipe_test` — compose-backed: Ingest persists the request + URI mapping, publishes to the process topic, and a re-ingest dedups to the same id. ✅ `bazel build //...`, `make fmt`.
behinddwalls
changed the base branch from
preetam/stovepipe-sourcecontrol
to
main
behinddwalls
force-pushed
the
preetam/stovepipe-ingest-integration
branch
from
6695ab7 to
8d1088f
Compare
behinddwalls
temporarily deployed
to
stack-rebase
GitHub Actions
Inactive
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.
Summary
Why?
Ingest was a thin stub: it minted a request id but never resolved the commit URI, persisted anything, or moved the request onto the pipeline, so
Request.URIstayed empty and nothing consumed the work. This makes ingest the real pipeline entry and adds the first internal queue contract so the pipeline can hand work to the next stage.What?
Ingest now resolves the queue's head URI via the SourceControl extension, dedups on the (queue, URI) pair, persists the Request and its URI mapping via storage, and publishes the request id to a new process stage over the messaging queue. Ingestion is idempotent: a re-reported head resolves to the already-minted request and nothing is published again. The URI mapping is claimed before the request row is written, so a lost race leaves no orphan row.
Adds the first internal proto message-queue contract under
stovepipe/core/messagequeue(proto3 + protojson, mirroringapi/runway/messagequeue): aProcessRequestpayload carrying the id, theTopicKeyProcessconstant, and the protojson glue, wired into the proto codegen (tool/proto,PROTO_PACKAGES). Per CLAUDE.md, internal contracts live under the domain'score/, notapi/, and the contract package owns both the payload and its topic keys.Adds a minimal
processconsumer (stovepipe/controller/process) that reloads the Request by id and logs it; a not-yet-visible request is retryable so redelivery converges. The build-strategy/ancestry logic the RFC assigns toprocessis deferred.Wires the example server (
example/stovepipe/server) into a MySQL storage + MySQL queue + fake SourceControl stack with the process consumer running, plus docker-compose (two databases) and a schema-init make target.Test Plan
✅
bazel test //stovepipe/...— contract round-trip + topic-key binding, ingest (happy/dedup/race/unknown-queue/infra-error paths), and process consumer unit tests.✅
bazel test //test/integration/stovepipe:stovepipe_test— compose-backed: Ingest persists the request + URI mapping, publishes to the process topic, and a re-ingest dedups to the same id.✅
bazel build //...,make fmt.Stack