Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions doc/rfc/stovepipe/project-status-api.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ Repository and project breakage degrees are independent projections. A degree is
| Request identity, baseline, and lifecycle | Durable request summary |
| Repository breakage degree | Repository validation fact |
| Project results | Implementation-defined project-result records |
| Project completion and pagination | Project-result completion record and cursor |
| Project completion and pagination | Completion timestamp in the durable request summary and cursor |

## Request Selection

Expand Down Expand Up @@ -72,7 +72,7 @@ Callers use `project_results_complete`, rather than the presence or absence of a

## Implementation

Both endpoints return durable lifecycle summaries, the whole-repository fact when present, and recorded facts for requested projects in paginated request order, or recorded project results by descending breakage degree then ascending project ID when none are requested. `project_results_complete` remains false until a durable completion projection is available. Missing facts are omitted; an explicitly recorded zero degree remains present. `updated_at_ms` is the maximum timestamp of the summary and the facts represented on the page.
Both endpoints return durable lifecycle summaries, the whole-repository fact when present, and recorded facts for requested projects in paginated request order, or recorded project results by descending breakage degree then ascending project ID when none are requested. `project_results_complete` is true when the durable request summary has a positive `project_results_completed_at_ms`, projected from the retained `project_facts_recorded` event. The record stage writes that event after all project facts and ordering mappings, including a successfully resolved empty set, and before publishing its terminal hook. The materializer persists the completion timestamp before the record stage continues, without changing lifecycle state or request version. Duplicate delivery repairs a failed projection using the original event timestamp, and subsequent state projections preserve completion. The API reads the summary before loading facts and includes its completion timestamp in `updated_at_ms`, without an additional event lookup. Missing completion remains false. Missing facts are omitted; an explicitly recorded zero degree remains present. `updated_at_ms` is the maximum timestamp of the summary and the facts represented on the page.

A dedicated ordering mapping makes listing a bounded primary-key range read, not a sort over all facts. Listing reads at most `page_size + 1` ordering mappings and hydrates only the returned facts, plus the cursor fact on continuations. The continuation token identifies the last project; its immutable fact supplies the degree needed to resume this ordering.

Expand Down
2 changes: 2 additions & 0 deletions doc/rfc/stovepipe/request-log.md
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,8 @@ For a Request transition, the controller:

Request creation, Build changes, and fact creation use the same source-write, log-write, dependent-publish ordering. A request-log outage can leave a source update visible, but it cannot allow dependent processing to move past an unrecorded transition.

The `project_facts_recorded` event also projects its retained timestamp into `RequestSummary.ProjectResultsCompletedAtMs`. This immutable fact is independent of lifecycle state and request version; state updates preserve it, and optimistic locking retries concurrent projection updates. Completion requires an existing lifecycle summary rather than fabricating state from the event. A failed projection fails persistence, and redelivery reconciles the retained occurrence before retrying the summary update. Other non-state events do not update the summary.

### Controller ownership

| Owner | Source operation and retained log | Retry repair |
Expand Down
8 changes: 4 additions & 4 deletions doc/rfc/stovepipe/steps/record.md
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describe

Phase 1 uses the states in [stovepipe/entity/request.go](../../../../stovepipe/entity/request.go). `record` runs *after* the Request is terminal: `buildsignal` projects the build's terminal status onto it as `succeeded`, `failed`, or `cancelled` (`RequestState.HasBuildOutcome()`), and only then publishes. So `record` reads an outcome and writes no state. `superseded` is terminal without an outcome.

Phase 2 broadens "complete" to "all planned facts recorded", which needs a marker this stage does not own; see [Completion marker: open](#completion-marker-open).
The `project_facts_recorded` event marks completion of the resolver’s result batch. A future pipeline with independent project builds would need to cover the entire planned set; see [Completion across independent project builds](#completion-across-independent-project-builds).

## Storage and queue contract

Expand Down Expand Up @@ -299,17 +299,17 @@ The rough idea is to track this with additional stores written during this stage

Neither fits `ValidationFactStore`. Facts are looked up by exact identity and URIs do not sort, while previous/next-green needs an ordered seek over positions, a different key shape. A genuinely needed reverse lookup getting its own first-class store is the established pattern here; `RequestURIStore` is the existing example (see [storage README](../../../../stovepipe/extension/storage/README.md#key-value-contract)).

### Completion marker: open
### Completion across independent project builds

`record` makes no `Request` write in Phase 1, so the stage has no durable marker saying it finished. `buildsignal` drives the Request terminal before `record` ever runs, leaving no non-terminal window to occupy.
`record` retains `project_facts_recorded` after all resolved project facts and ordering mappings are durable, including an empty batch, and before publishing its terminal hook. The materializer projects its original timestamp into the request summary before the stage continues, and project status APIs read completion from that summary. A projection failure fails the delivery; replay repairs the summary before the terminal hook. If a future pipeline fans a request out into independent project builds, completion must cover that entire planned set before emitting the marker. Request terminality alone is insufficient because it precedes project recording.

### Supporting re-run of the same URI

Widen the key with the `RequestID` the facts already record, so each attempt is its own immutable row, and add a pointer store from `(queue, uri, project)` to the authoritative attempt. Advance to a newer attempt unless the current one is green, since a green build proved the code passed and a later failure only proves the build is non-deterministic.

### Phase 2

- **Pipeline**: `record` publishes the request id onward to `analyze`, for green and not-green facts alike, because a failed build is when project attribution matters most. An earlier draft had `record` retarget the Request from `processing` to `analyzing`, with `analyze` owning the terminal transition. That no longer fits, because the Request is already terminal before `record` runs, so tracking "all facts recorded" belongs to the `analyze` design (see [Completion marker: open](#completion-marker-open)). The message onward carries the request id and queue name, exactly as `record`'s own does, and the consumer stays idempotent.
- **Pipeline**: `record` publishes the request id onward to `analyze`, for green and not-green facts alike, because a failed build is when project attribution matters most. An earlier draft had `record` retarget the Request from `processing` to `analyzing`, with `analyze` owning the terminal transition. That no longer fits, because the Request is already terminal before `record` runs, so tracking "all facts recorded" belongs to the `analyze` design (see [Completion across independent project builds](#completion-across-independent-project-builds)). The message onward carries the request id and queue name, exactly as `record`'s own does, and the consumer stays idempotent.
- **Project facts**: `record` runs the same create-fact-then-notify flow. The fact is keyed by the stable project id carried on the per-project signal, and one `validation.project.recorded` event is published per project identity (see [What the](#what-the-type-carries) `type` [carries](#what-the-type-carries)). Each per-project signal needs its own message id (see [Storage and queue contract](#storage-and-queue-contract)). The Queue's `last_green_uri` describes the whole repository and stays untouched.

Deciding project identity, retrieving the target graph, tracking completion, and defining intermediate degrees belong to `analyze.md`.
3 changes: 2 additions & 1 deletion stovepipe/controller/project_status.go
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,8 @@ func (c *projectStatusController) GetProjectStatus(ctx context.Context, req enti
return entity.GetProjectStatusResult{}, &ProjectStatusConsistencyError{message: "project status selector disagrees with stored request summary"}
}
result.RequestSummary = requestSummary
result.UpdatedAtMs = requestSummary.StateTimestampMs
result.ProjectResultsComplete = requestSummary.ProjectResultsCompletedAtMs > 0
result.UpdatedAtMs = max(requestSummary.StateTimestampMs, requestSummary.ProjectResultsCompletedAtMs)
factStore := store.GetValidationFactStore()
fact, err := factStore.Get(ctx, requestSummary.URI, "")
if err != nil {
Expand Down
63 changes: 63 additions & 0 deletions stovepipe/controller/project_status_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -590,3 +590,66 @@ func TestGetProjectStatusRejectsInconsistentRequest(t *testing.T) {
assert.True(t, IsProjectStatusConsistency(err))
assert.False(t, errors.Is(err, storage.ErrNotFound))
}

func TestGetProjectStatusCompletion(t *testing.T) {
for _, tt := range []struct {
name string
completedAt int64
stateAt int64
empty bool
updatedAt int64
}{
{name: "terminal request still recording", stateAt: 10, updatedAt: 20},
{name: "completed projects", completedAt: 30, stateAt: 10, updatedAt: 30},
{name: "completed empty set", completedAt: 30, stateAt: 10, empty: true, updatedAt: 30},
{name: "newer lifecycle timestamp", completedAt: 30, stateAt: 40, updatedAt: 40},
{name: "newer fact timestamp", completedAt: 15, stateAt: 10, updatedAt: 20},
} {
for _, byURI := range []bool{false, true} {
selector := "ID"
if byURI {
selector = "URI"
}
t.Run(tt.name+"/"+selector, func(t *testing.T) {
ctrl := gomock.NewController(t)
factory := storagemock.NewMockFactory(ctrl)
store := storagemock.NewMockStorage(ctrl)
summaries := storagemock.NewMockRequestSummaryStore(ctrl)
facts := storagemock.NewMockValidationFactStore(ctrl)
summary := entity.RequestSummary{
Queue: projectStatusQueue, RequestID: projectStatusID, URI: projectStatusURI,
State: entity.RequestStateFailed, StateTimestampMs: tt.stateAt,
ProjectResultsCompletedAtMs: tt.completedAt,
}
factory.EXPECT().For(storage.Config{QueueName: projectStatusQueue}).Return(store, nil)
req := entity.GetProjectStatusRequest{Queue: projectStatusQueue, RequestID: projectStatusID, Projects: []string{"p", "missing"}, PageSize: 1}
if byURI {
req.RequestID = ""
req.ChangeURI = projectStatusURI
uris := storagemock.NewMockRequestURIStore(ctrl)
store.EXPECT().GetRequestURIStore().Return(uris)
uris.EXPECT().GetIDByURI(gomock.Any(), projectStatusURI).Return(projectStatusID, nil)
}
store.EXPECT().GetRequestSummaryStore().Return(summaries)
readSummary := summaries.EXPECT().Get(gomock.Any(), projectStatusID).Return(summary, nil)
store.EXPECT().GetValidationFactStore().Return(facts).After(readSummary)
facts.EXPECT().Get(gomock.Any(), projectStatusURI, "").Return(entity.ValidationFact{}, storage.ErrNotFound)
if tt.empty {
facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(entity.ValidationFact{}, storage.ErrNotFound)
} else {
facts.EXPECT().Get(gomock.Any(), projectStatusURI, "p").Return(entity.ValidationFact{URI: projectStatusURI, RequestID: projectStatusID, Project: "p", Degree: 1, CreatedAt: 20}, nil)
}
result, err := NewProjectStatusController(zap.NewNop().Sugar(), tally.NoopScope, factory).GetProjectStatus(context.Background(), req)
require.NoError(t, err)
assert.Equal(t, tt.completedAt > 0, result.ProjectResultsComplete)
assert.Equal(t, tt.updatedAt, result.UpdatedAtMs)
if tt.empty {
assert.Empty(t, result.ProjectValidationFacts)
} else {
assert.Len(t, result.ProjectValidationFacts, 1)
}
assert.NotEmpty(t, result.NextPageToken)
})
}
}
}
3 changes: 0 additions & 3 deletions stovepipe/controller/record/record.go
Original file line number Diff line number Diff line change
Expand Up @@ -226,9 +226,6 @@ func (c *Controller) persistProjectFacts(ctx context.Context, store storage.Stor
return fmt.Errorf("failed to index project fact for request %q and project %q: %w", request.ID, fact.Project, err)
}
}
if len(results) == 0 {
return nil
}
return c.persistProjectFactsRecordedLog(ctx, store, request, len(results))
}

Expand Down
63 changes: 61 additions & 2 deletions stovepipe/controller/record/record_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -163,11 +163,11 @@ func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consu
m.store.EXPECT().GetProjectFactByDegreeStore().Return(m.indexStore).AnyTimes()
m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(
func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
assert.Equal(t, entity.RequestEventValidationFactRecorded, log.Event)
assert.Contains(t, []entity.RequestEvent{entity.RequestEventValidationFactRecorded, entity.RequestEventProjectFactsRecorded}, log.Event)
assert.Empty(t, log.State)
assert.Zero(t, log.RequestVersion)
assert.Empty(t, log.OutcomeReason)
assert.NotEmpty(t, log.Metadata[requestlog.MetadataKeyFactDegree])
assert.NotEmpty(t, log.Metadata)
return nil
},
).AnyTimes()
Expand Down Expand Up @@ -405,6 +405,64 @@ func TestRecordProjectFacts_PassesEmptyTerminalBuild(t *testing.T) {
require.NoError(t, err)
}

func TestProcessRecordsEmptyProjectCompletion(t *testing.T) {
resolveErr := errors.New("project resolution unavailable")
writeErr := errors.New("completion write unavailable")
for _, tt := range []struct {
name string
results []projectstatus.Result
resolveErr error
writeErr error
}{
{name: "nil results"},
{name: "empty results", results: []projectstatus.Result{}},
{name: "resolution failed", resolveErr: resolveErr},
{name: "completion write failed", writeErr: writeErr},
} {
t.Run(tt.name, func(t *testing.T) {
ctrl := gomock.NewController(t)
c, m := newController(t, ctrl)
request := requestWithState(entity.RequestStateFailed)
m.reqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil)
factory := projectstatusmock.NewMockResolverFactory(ctrl)
resolver := projectstatusmock.NewMockResolver(ctrl)
c.projectStatusResolverFactory = factory
factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: testQueue}).Return(resolver, nil)
resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return(tt.results, tt.resolveErr)
materializer := requestlogmock.NewMockMaterializer(ctrl)
c.materializer = materializer
if tt.resolveErr == nil {
m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
gomock.InOrder(
materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(
func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
assert.Equal(t, entity.RequestEventValidationFactRecorded, log.Event)
return nil
}),
materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(
func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
assert.Equal(t, "event/project_facts_recorded/all", log.ID)
assert.Equal(t, "0", log.Metadata[requestlog.MetadataKeyProjectFactCount])
return tt.writeErr
}),
)
}
err := c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, request.ID)))
switch {
case tt.resolveErr != nil:
require.ErrorIs(t, err, tt.resolveErr)
assert.Empty(t, m.hooks.events)
case tt.writeErr != nil:
require.ErrorIs(t, err, tt.writeErr)
assert.Empty(t, m.hooks.events)
default:
require.NoError(t, err)
assert.Len(t, m.hooks.events, 1)
}
})
}
}

func TestProcess_RejectsInvalidProjectStatusResultsBeforeWritingFacts(t *testing.T) {
ctrl := gomock.NewController(t)
c, m := newController(t, ctrl)
Expand Down Expand Up @@ -538,6 +596,7 @@ func TestProcess_RecordsFactEventBeforeDerivedWork(t *testing.T) {
m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).
Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil)
m.sourceControl.EXPECT().Promote(gomock.Any(), testURI).Return(nil)
eventMaterializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil)

require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))))
}
Expand Down
2 changes: 1 addition & 1 deletion stovepipe/core/requestlog/materializer.go
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l
return err
}

if retained.State == entity.RequestStateUnknown {
if retained.State == entity.RequestStateUnknown && retained.Event != entity.RequestEventProjectFactsRecorded {
return nil
}
if err := materializeRequestProjections(ctx, stores, retained); err != nil {
Expand Down
17 changes: 17 additions & 0 deletions stovepipe/core/requestlog/materializer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -383,3 +383,20 @@ func TestMaterializerMetricsIncludeContextTags(t *testing.T) {
require.True(t, ok)
assert.EqualValues(t, 1, counter.Value())
}

func TestMaterializerRetainsOtherEventsWithoutProjectingSummary(t *testing.T) {
for _, event := range []entity.RequestEvent{
entity.RequestEventBuildTriggered,
entity.RequestEventBuildFinished,
entity.RequestEventValidationFactRecorded,
entity.RequestEventRecordAbandoned,
entity.RequestEventBuildAnalysisAbandoned,
} {
t.Run(string(event), func(t *testing.T) {
materializer, stores, logs, _, _ := newTestMaterializer(t)
log := NewRequestEventLog(entity.Request{ID: testRequestID, Queue: testQueue}, event, "test", nil)
logs.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
require.NoError(t, materializer.PersistLog(context.Background(), stores, log))
})
}
}
21 changes: 16 additions & 5 deletions stovepipe/core/requestlog/request_summary.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,9 @@ func materializeRequestSummary(ctx context.Context, stores storage.Storage, log
for {
currentSummary, err := summaryStore.Get(ctx, log.RequestID)
if storage.IsNotFound(err) {
if log.State == entity.RequestStateUnknown {
return entity.RequestSummary{}, fmt.Errorf("request summary required to project event request_id=%q: %w", log.RequestID, err)
}
initial, createErr := createInitialRequestSummary(ctx, stores.GetRequestStore(), summaryStore, log)
if errors.Is(createErr, storage.ErrAlreadyExists) {
continue
Expand Down Expand Up @@ -113,17 +116,25 @@ func projectRequestSummaryFacts(summary entity.RequestSummary, log entity.Reques
log.RequestID, summary.RequestID, log.Queue, summary.Queue,
)
}
if log.State != entity.RequestStateAccepted {
return summary, nil
if log.Event == entity.RequestEventProjectFactsRecorded {
if log.TimestampMs <= 0 || (summary.ProjectResultsCompletedAtMs != 0 && summary.ProjectResultsCompletedAtMs != log.TimestampMs) {
return entity.RequestSummary{}, fmt.Errorf("request summary completion time conflicts with retained log request_id=%q", log.RequestID)
}
summary.ProjectResultsCompletedAtMs = log.TimestampMs
}
if summary.AcceptedAtMs != 0 && summary.AcceptedAtMs != log.TimestampMs {
return entity.RequestSummary{}, fmt.Errorf("request summary acceptance time conflicts with retained log request_id=%q", log.RequestID)
if log.State == entity.RequestStateAccepted {
if summary.AcceptedAtMs != 0 && summary.AcceptedAtMs != log.TimestampMs {
return entity.RequestSummary{}, fmt.Errorf("request summary acceptance time conflicts with retained log request_id=%q", log.RequestID)
}
summary.AcceptedAtMs = log.TimestampMs
}
summary.AcceptedAtMs = log.TimestampMs
return summary, nil
}

func projectRequestSummaryState(summary entity.RequestSummary, log entity.RequestLog) (entity.RequestSummary, error) {
if log.State == entity.RequestStateUnknown {
return summary, nil
}
if log.RequestVersion < summary.RequestVersion {
return summary, nil
}
Expand Down
Loading
Loading