Skip to content

Commit 3a326f4

Browse files
committed
fix(server): stop dropping step reports the timeline path carries
Four defects in the previous three commits. The record filter whitelisted `Task`, but `type` is optional on the wire and this same file already treats `Step` as a step 120 lines above, so every untyped or `Step` record lost its conclusion. Container records are excluded instead; untyped ones fall back to the parent-id signal the annotation path uses. The old test sent only `Task` records, so it asserted the allow-list back at itself; it now sends all three shapes. The revision column moved into its own migration. Editing an applied version left any database already at 4 without the column, failing every step write on an undefined column. The restore paths never read `revision`, so the counter resumed at zero and the guard discarded the first writes after a restart -- silently, since memory stays authoritative until the next one. Seeded from the persisted maximum, covered by a two-restart test. The manifest rebuild matched queue rows by (run, job) rather than request id, attaching the newest queued message to an older attempt and rebuilding it with step ids its log blobs are not named after. Same attempt-versus-job confusion as 43c6301. Reported on #212 by Cubic.
1 parent 22fba73 commit 3a326f4

6 files changed

Lines changed: 186 additions & 36 deletions

File tree

‎crates/preloop-runner-server/src/lib_tests.rs‎

Lines changed: 107 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2912,28 +2912,40 @@ async fn timeline_path_orders_synthetic_steps_and_skips_the_job_record() {
29122912

29132913
// Setup started before the declared steps, and the job record is sent
29142914
// alongside them exactly as the official runner does.
2915+
//
2916+
// The three step records deliberately use the three shapes that appear on
2917+
// the wire: `Task` (official runner), `Step` (which this file's annotation
2918+
// path already recognises), and no `type` at all (the field is optional).
2919+
// An allow-list of one type passes a test that only sends that type while
2920+
// silently dropping the others' conclusions.
29152921
let setup_id = uuid::Uuid::new_v4().to_string();
2916-
let record = |id: &str, kind: &str, name: &str, start: &str| {
2917-
json!({
2922+
let job_record_id = uuid::Uuid::new_v4().to_string();
2923+
let record = |id: &str, kind: Option<&str>, name: &str, start: &str| {
2924+
let mut value = json!({
29182925
"id": id,
2919-
"type": kind,
29202926
"name": name,
29212927
"displayName": name,
29222928
"state": "completed",
29232929
"result": "succeeded",
29242930
"startTime": start,
2925-
})
2931+
});
2932+
match kind {
2933+
Some(kind) => value["type"] = json!(kind),
2934+
// Untyped records are identified as steps by their parent.
2935+
None => value["parentId"] = json!(job_record_id),
2936+
}
2937+
value
29262938
};
29272939
let response = request_json(
29282940
&app,
29292941
Method::PATCH,
29302942
&format!("/_apis/v1/Timeline/scope/actions/{plan_id}/{timeline_id}"),
29312943
json!({"count": 5, "value": [
2932-
record(&uuid::Uuid::new_v4().to_string(), "Job", "build", "2026-01-01T00:00:00Z"),
2933-
record(&ids[0], "Task", "Run echo one", "2026-01-01T00:00:02Z"),
2934-
record(&ids[1], "Task", "Run echo two", "2026-01-01T00:00:03Z"),
2935-
record(&ids[2], "Task", "Run echo three", "2026-01-01T00:00:04Z"),
2936-
record(&setup_id, "Task", "Set up job", "2026-01-01T00:00:01Z"),
2944+
record(&job_record_id, Some("Job"), "build", "2026-01-01T00:00:00Z"),
2945+
record(&ids[0], Some("Task"), "Run echo one", "2026-01-01T00:00:02Z"),
2946+
record(&ids[1], Some("Step"), "Run echo two", "2026-01-01T00:00:03Z"),
2947+
record(&ids[2], None, "Run echo three", "2026-01-01T00:00:04Z"),
2948+
record(&setup_id, Some("Task"), "Set up job", "2026-01-01T00:00:01Z"),
29372949
]}),
29382950
)
29392951
.await;
@@ -3185,6 +3197,92 @@ async fn step_manifests_survive_a_restart() {
31853197
);
31863198
}
31873199

3200+
/// A step report made after a restart is still persisted.
3201+
///
3202+
/// The out-of-order write guard compares an in-memory revision against the
3203+
/// persisted one, so the counter has to resume above what is on disk. Left at
3204+
/// zero it hands every post-restart write a revision the stored rows already
3205+
/// exceed, and the upsert discards them — invisibly, because memory stays
3206+
/// authoritative until the next restart drops the conclusion.
3207+
#[tokio::test]
3208+
async fn step_reports_after_a_restart_are_persisted() {
3209+
let temp = tempfile::tempdir().unwrap();
3210+
let report = |app: axum::Router, plan_id: String, agent_job_id: String, id: String, n: i64| async move {
3211+
let response = request_json(
3212+
&app,
3213+
Method::POST,
3214+
"/twirp/github.actions.results.api.v1.WorkflowStepUpdateService/WorkflowStepsUpdate",
3215+
json!({
3216+
"workflow_run_backend_id": plan_id,
3217+
"workflow_job_run_backend_id": agent_job_id,
3218+
"steps": [{
3219+
"external_id": id,
3220+
"number": n,
3221+
"name": "Run echo one",
3222+
"status": 6,
3223+
"conclusion": 2
3224+
}]
3225+
}),
3226+
)
3227+
.await;
3228+
assert_eq!(response["ok"], true);
3229+
};
3230+
3231+
let (plan_id, agent_job_id, ids) = {
3232+
let state = AppState::new(temp.path().to_path_buf()).await.unwrap();
3233+
let app = app(state.clone(), CancellationToken::new());
3234+
let (run_id, jobs) = three_step_run_for_log_filters(&app, &state).await;
3235+
let ids = workflow_step_ids(&state, run_id, "build").await;
3236+
let (plan_id, agent_job_id) = (jobs[0].1.clone(), jobs[0].2.clone());
3237+
// Two reports before the restart, so the persisted revision is above
3238+
// the value a fresh counter would hand out first.
3239+
report(
3240+
app.clone(),
3241+
plan_id.clone(),
3242+
agent_job_id.clone(),
3243+
ids[0].clone(),
3244+
2,
3245+
)
3246+
.await;
3247+
report(
3248+
app.clone(),
3249+
plan_id.clone(),
3250+
agent_job_id.clone(),
3251+
ids[1].clone(),
3252+
3,
3253+
)
3254+
.await;
3255+
(plan_id, agent_job_id, ids)
3256+
};
3257+
3258+
{
3259+
let state = AppState::new(temp.path().to_path_buf()).await.unwrap();
3260+
let app = app(state.clone(), CancellationToken::new());
3261+
report(
3262+
app,
3263+
plan_id.clone(),
3264+
agent_job_id.clone(),
3265+
ids[2].clone(),
3266+
4,
3267+
)
3268+
.await;
3269+
}
3270+
3271+
// Only a second restart can tell whether that write reached the store.
3272+
let state = AppState::new(temp.path().to_path_buf()).await.unwrap();
3273+
let inner = state.inner.lock().await;
3274+
let records = &inner.job_steps[&agent_job_id.parse::<uuid::Uuid>().unwrap()];
3275+
let reported = records
3276+
.iter()
3277+
.find(|step| step.id == ids[2])
3278+
.expect("the post-restart step must be restored");
3279+
assert_eq!(
3280+
reported.conclusion, "success",
3281+
"a report made after a restart must survive the next one"
3282+
);
3283+
assert_eq!(reported.runner_number, Some(4));
3284+
}
3285+
31883286
/// Every surface showing a whole step list uses execution order.
31893287
///
31903288
/// The stored manifest is seeded with declared steps and then appends

‎crates/preloop-runner-server/src/runs.rs‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1084,11 +1084,13 @@ pub(crate) async fn submit_run_inner(
10841084
// message we just built rather than from whatever order the
10851085
// runner's log blobs happen to land in.
10861086
//
1087-
// Not persisted here. Only a runner report writes rows, and an
1088-
// attempt that was dispatched but never reported has its
1089-
// manifest rebuilt at startup from the persisted request
1090-
// message — the same source it was built from — so a restart
1091-
// in that window keeps its declared steps.
1087+
// Not persisted here. Rows are written by a runner report, a
1088+
// job completion, or a full snapshot that happens to flush
1089+
// them; none of those has necessarily run when a dispatched
1090+
// attempt is interrupted, so its manifest is rebuilt at startup
1091+
// from the persisted request message — the same source it was
1092+
// built from — and a restart in that window keeps its declared
1093+
// steps.
10921094
inner.job_steps.insert(
10931095
job_request.agent_job_id,
10941096
StepRecord::manifest(&agent_msg.steps),

‎crates/preloop-runner-server/src/state.rs‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -778,7 +778,13 @@ impl AppState {
778778
.iter()
779779
.chain(recovered.pending_jobs.iter())
780780
.chain(recovered.concurrency_blocked.iter())
781-
.find(|job| job.run_id == record.run_id && job.job_id == record.job_id)
781+
// Keyed by request id, not by (run, job): a
782+
// re-dispatch leaves several requests for one
783+
// logical job, and matching the pair attaches the
784+
// newest queued message to an older attempt —
785+
// rebuilding it with the wrong `TaskStep` ids, so
786+
// its `step-<id>.txt` blobs stop resolving.
787+
.find(|job| job.message.request_id == record.request_id)
782788
.map(|job| job.message.steps.as_slice())
783789
})?;
784790
let manifest = crate::models::StepRecord::manifest(steps);

‎crates/preloop-runner-server/src/store.rs‎

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1437,7 +1437,8 @@ impl SqliteStore {
14371437
// guessing, so the run's step history has to come back here.
14381438
let mut step_stmt = connection.prepare(
14391439
"SELECT agent_job_id, step_id, kind, workflow_index, runner_number,
1440-
context_name, name_blob, conclusion, started_at_us, finished_at_us
1440+
context_name, name_blob, conclusion, started_at_us, finished_at_us,
1441+
revision
14411442
FROM job_steps
14421443
ORDER BY agent_job_id, COALESCE(runner_number, 2147483647),
14431444
COALESCE(workflow_index, 2147483647), step_id",
@@ -1454,6 +1455,7 @@ impl SqliteStore {
14541455
row.get::<_, String>(7)?,
14551456
row.get::<_, Option<i64>>(8)?,
14561457
row.get::<_, Option<i64>>(9)?,
1458+
row.get::<_, i64>(10)?,
14571459
))
14581460
})? {
14591461
let (
@@ -1467,6 +1469,7 @@ impl SqliteStore {
14671469
conclusion,
14681470
started_at_us,
14691471
finished_at_us,
1472+
revision,
14701473
) = row?;
14711474
let Ok(agent_job_id) = agent_job_id.parse::<uuid::Uuid>() else {
14721475
tracing::warn!(%agent_job_id, "dropping step row with an unparseable attempt id");
@@ -1479,6 +1482,12 @@ impl SqliteStore {
14791482
continue;
14801483
}
14811484
};
1485+
// The guard compares against the persisted revision, so the
1486+
// counter has to resume above it. Left at zero, the first writes
1487+
// after a restart carry revisions the rows already exceed and are
1488+
// silently discarded.
1489+
let seen = inner.job_steps_revision.entry(agent_job_id).or_insert(0);
1490+
*seen = (*seen).max(revision.max(0) as u64);
14821491
inner
14831492
.job_steps
14841493
.entry(agent_job_id)
@@ -2406,19 +2415,28 @@ const MIGRATIONS: &[(u32, &str, &str)] = &[
24062415
conclusion TEXT NOT NULL,
24072416
started_at_us INTEGER,
24082417
finished_at_us INTEGER,
2409-
-- Monotonic per attempt, bumped on every in-memory mutation. Two
2410-
-- reports for one attempt snapshot under the lock and then commit
2411-
-- outside it, so they can commit out of order; the upsert refuses to
2412-
-- move a row backwards rather than letting the older snapshot
2413-
-- overwrite the newer conclusions.
2414-
revision INTEGER NOT NULL DEFAULT 0,
24152418
PRIMARY KEY (agent_job_id, step_id)
24162419
) STRICT, WITHOUT ROWID;
24172420
24182421
CREATE INDEX IF NOT EXISTS job_steps_order_idx
24192422
ON job_steps (agent_job_id, kind, workflow_index);
24202423
"#,
24212424
),
2425+
(
2426+
5,
2427+
"job-steps-revision",
2428+
// Monotonic per attempt, bumped on every in-memory mutation of the
2429+
// manifest. Reconciliation snapshots under the state lock and writes
2430+
// after releasing it, so two reports for one attempt can commit out of
2431+
// order; the upsert compares this and refuses to move a row backwards.
2432+
//
2433+
// Its own version rather than an edit to 4: a database already at 4
2434+
// from an earlier build of this branch would skip the change entirely
2435+
// and fail every step write with an undefined column.
2436+
r#"
2437+
ALTER TABLE job_steps ADD COLUMN revision INTEGER NOT NULL DEFAULT 0;
2438+
"#,
2439+
),
24222440
];
24232441

24242442
pub(crate) fn now_us() -> i64 {

‎crates/preloop-runner-server/src/store_pg.rs‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -643,7 +643,8 @@ impl Store for PgStore {
643643
let rows = client
644644
.query(
645645
"SELECT agent_job_id, step_id, kind, workflow_index, runner_number,
646-
context_name, name_blob, conclusion, started_at_us, finished_at_us
646+
context_name, name_blob, conclusion, started_at_us, finished_at_us,
647+
revision
647648
FROM job_steps
648649
ORDER BY agent_job_id, COALESCE(runner_number, 2147483647),
649650
COALESCE(workflow_index, 2147483647), step_id",
@@ -665,13 +666,18 @@ impl Store for PgStore {
665666
let conclusion: String = row.get(7);
666667
let started_at_us: Option<i64> = row.get(8);
667668
let finished_at_us: Option<i64> = row.get(9);
669+
let revision: i64 = row.get(10);
668670
let name = match self.cipher.unseal(&name_blob) {
669671
Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
670672
Err(error) => {
671673
tracing::warn!(%error, %step_id, "dropping step row with an unreadable name");
672674
continue;
673675
}
674676
};
677+
// See the SQLite twin: the counter must resume above the
678+
// persisted revision or the guard discards post-restart writes.
679+
let seen = inner.job_steps_revision.entry(agent_job_id).or_insert(0);
680+
*seen = (*seen).max(revision.max(0) as u64);
675681
inner
676682
.job_steps
677683
.entry(agent_job_id)
@@ -1288,14 +1294,23 @@ const MIGRATIONS: &[(u32, &str, &str)] = &[
12881294
conclusion TEXT NOT NULL,
12891295
started_at_us BIGINT,
12901296
finished_at_us BIGINT,
1291-
-- See the SQLite twin: guards against two reports for one attempt
1292-
-- committing out of order.
1293-
revision BIGINT NOT NULL DEFAULT 0,
12941297
PRIMARY KEY (agent_job_id, step_id)
12951298
);
12961299
12971300
CREATE INDEX IF NOT EXISTS job_steps_order_idx
12981301
ON job_steps (agent_job_id, kind, workflow_index);
12991302
"#,
13001303
),
1304+
(
1305+
5,
1306+
"job-steps-revision",
1307+
// See the SQLite twin: guards two reports for one attempt committing
1308+
// out of order. Its own version rather than an edit to 4, because a
1309+
// database already at 4 from an earlier build of this branch would
1310+
// skip the change and fail every step write on an undefined column.
1311+
r#"
1312+
ALTER TABLE job_steps
1313+
ADD COLUMN IF NOT EXISTS revision BIGINT NOT NULL DEFAULT 0;
1314+
"#,
1315+
),
13011316
];

‎crates/preloop-runner-server/src/timeline_logs.rs‎

Lines changed: 21 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -156,16 +156,27 @@ pub(crate) async fn patch_timeline_records(
156156
let Some(name) = &record.display_name else {
157157
continue;
158158
};
159-
// A PATCH carries the job's own record alongside its
160-
// steps. Its id is a UUID and `job_id.0` is the workflow
161-
// job key, so the id comparison below never matched it and
162-
// the job was reconciled in as a synthetic step — showing
163-
// up in run projections as an extra step named after the
164-
// job, and persisted there. Only task records are steps.
165-
if !matches!(record.record_type, Some(azdo::TimelineRecordType::Task)) {
166-
continue;
167-
}
168-
if record.id.to_string() == job_id.0 {
159+
// A PATCH carries the job's own record alongside its steps,
160+
// and reconciling that record in produced a phantom step
161+
// named after the job. Exclude container records rather
162+
// than whitelisting one step type: `type` is optional on
163+
// the wire, the official runner sends `Task` and this file
164+
// already treats `Step` as a step above, so an allow-list
165+
// silently drops real conclusions.
166+
let is_step = match record.record_type {
167+
Some(azdo::TimelineRecordType::Task | azdo::TimelineRecordType::Step) => {
168+
true
169+
}
170+
Some(
171+
azdo::TimelineRecordType::Job
172+
| azdo::TimelineRecordType::Phase
173+
| azdo::TimelineRecordType::Stage,
174+
) => false,
175+
// Untyped: a step always hangs off the job record, which
176+
// is the same signal the annotation path above uses.
177+
None => record.parent_id.is_some(),
178+
};
179+
if !is_step {
169180
continue;
170181
}
171182

0 commit comments

Comments
 (0)