Repository navigation
Native: remove additional redundant transition reads without stale wait ordering #1077
Description
Current owner and source work — November 9, 2026
General owns this P0 while the Server agent stays Off; all original WIP refs are preserved. The seven #1063/#1099 reductions and the later #1068/#1075 work are already released in Native 3.3.2. This issue owns the additional cuts, with no duplicate census or wait-ranking shortcut.
Draft PR #1115 now implements one independent cut: remove the successful activity completion's redundant row reread while preserving its full history snapshot. Head 4613b8adf31f67f593080c5988e81b41a6f7a993, base 5bde7ed3, tree a9c6abef; four files. One normal source train is running; it is the first compile/test execution. Controls compare the full snapshot to a stored-row read, retain prior-attempt/progress/legacy-nullable fields and stale repeats, check cross-Store heartbeat serialization and count the removed SELECT. No local build, runtime experiment, merge or speedup is claimed.
Next: inspect the complete original CI attempt, correct any source/test failures without cancelling live lanes, then qualify this cut through ordinary integration and a published patch. The remaining independent cuts stay open for separate source review and causal controls; the supplied cumulative 0005/0006 references are not applied wholesale. Preserve unconditional chronological wait ranking, backend errors, original IDs, history budgets and rollback state.
Original requirement and evidence
External report topic 23, accepted for P0 intake at the user's direction. The proposal, timings and supplied test results are unverified reference evidence. No patch was applied or executed during intake.
Implementation owner: general. The owning agent drives source, release and applicable published-artifact verification. Existing release, operating and credential gates remain in force.
Problem and proposed scope
The report supplies additional transition cuts: query-poll cache operations in the existing transaction; compact remembered workflow-lease rows with history reconstruction; start payload joined with claim/run data; unused activity wait-clock reads; folding already-delivered history into summary scans; removing activity read-back; reusing probed run status and authored counters; and omitting impossible prior-attempt expiry.
The projection census overlaps the existing P0 server#1063. Keep that item in its existing home and deduplicate any other overlapping cut there; this issue owns the additional source delta and its rollback/order controls.
Acceptance
- Review and qualify each independent cut against maintained source. Stored rows, answers, fairness, history budgets, published millisecond ties and original IDs must remain equivalent.
- Preserve backend errors, committed visibility, idempotency and all savepoint cache invalidation. Rebuilt poll history must remain bounded and correlated to the original lease/attempt.
- Preserve unconditional chronological wait ranking when another projector's converging writes can move positions. The supplied 0005 stage is known wrong in this respect; retain 0006's correction and causal negatives, including current_wait_order_uses_published_millisecond_ties_and_original_ids and wait_positions_are_chronological_and_written_only_when_a_wait_moves.
- Preserve the actual server cancellation-policy contract instead of weakening it to make the supplied snapshot test pass.
- Relevant suites must pass on all three databases; retain genuine base/candidate failures and statement-count evidence. The earlier 14.51→12.61-second claim includes the wrong shortcut and is not a valid gain. The corrected result is within the reported session drift.
Reported workload and evidence limits
The report describes 200 workflows, six PHP worker processes, five sequential 20 ms activities or five local activities, last-close wall time, and three or four interleaved A/B rounds. MySQL session-to-session drift is reported as high as 11%. Stock database durability, network placement and the absence of deferred projections are reporter conditions, not independently verified facts. The headline run covers topics 1–24; topic 25 has only a MySQL five-steps A/B. No speedup, production readiness or suite pass is accepted from this intake.
Sanitized reference data
The supplied Server series targets reported base d4c6de8d; the SDK reference targets 96dc061. Maintained heads were independently read as Server c3bbfa5 and SDK c87eae37. These cumulative references are not current-main patches or maintained qualification.
0005-Cut-more-statements-from-each-workflow-and-activity-.patch— 103,089 bytes / SHA-2563095223e9455381ff492a08b7daf5651ce94aa138bff00cecaf3bc2bdfdc2306.0006-Rank-a-run-s-waits-on-every-transition-again-fix-the.patch— 8,143 bytes / SHA-2564d1d1bc2e7e150495a6109f8c4ee0b2d13e65da67a7c8add9c0ef4c46f8fc090.
Bench-only Server0001 is retained as diagnostic context, not a separate production change. Complete sanitized diff references, shared bundle identity and source-review findings are linked in the intake index and the issue comments. Original mail headers, author/date metadata, application commands and runner instructions are omitted.
Activity
rmcdaniel commentedon Nov 6, 2026
Sanitized external reference for topic 23:
0005-Cut-more-statements-from-each-workflow-and-activity-.patch, part 1/3. Complete sanitized diff: 103,089 bytes / SHA-2563095223e9455381ff492a08b7daf5651ce94aa138bff00cecaf3bc2bdfdc2306. Ordered parts preserve the complete diff bytes; none was applied or executed. The patch is a cumulative older-source draft, not maintained implementation or qualification. Mail headers/author/date/message/application instructions were removed.The supplied 0005 stage includes the known stale wait-ranking shortcut. It is not an acceptable production candidate by itself; retain and independently qualify correction0006 and the negative controls.
rmcdaniel commentedon Nov 6, 2026
Sanitized external reference for topic 23:
0005-Cut-more-statements-from-each-workflow-and-activity-.patch, part 2/3. Complete sanitized diff: 103,089 bytes / SHA-2563095223e9455381ff492a08b7daf5651ce94aa138bff00cecaf3bc2bdfdc2306. Ordered parts preserve the complete diff bytes; none was applied or executed. The patch is a cumulative older-source draft, not maintained implementation or qualification. Mail headers/author/date/message/application instructions were removed.The supplied 0005 stage includes the known stale wait-ranking shortcut. It is not an acceptable production candidate by itself; retain and independently qualify correction0006 and the negative controls.
recorded_at, ) .await?; } else if command["type"] == "start_timer" { - Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Timers, &limits, task_id, attempt).await?; + Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Timers, &limits, task_id, attempt, None).await?; sequence += 1; let metadata = super::selections::metadata(command, "timer")?; super::selections::check_sequence(&metadata, sequence)?; @@ -4134,17 +4267,17 @@ where self.claim_external_in(&mut tx, &prepared.reference, &prepared.uri) .await?; } + let closed_at = now(); Self::query("UPDATE activity_executions SET status='completed',result=$1,exception=NULL,closed_at=$2 WHERE id=$3") - .bind(&result).bind(DB::bind_time(now())).bind(&activity_id).execute(&mut *tx).await?; - let activity = Self::query("SELECT * FROM activity_executions WHERE id=$1") - .bind(&activity_id) - .fetch_one(&mut *tx) - .await?; - let mut payload = Self::activity_payload(&activity)?; - let metadata = super::selections::activity_metadata(DB::optional_document_row( - &activity, - "parallel_group_path", - )?)?; + .bind(&result).bind(DB::bind_time(closed_at)).bind(&activity_id).execute(&mut *tx).await?; + // The row as the UPDATE leaves it: its time is whole milliseconds, + // which every backend stores exactly. + let mut activity = ActivityRecord::read::<DB>(&activity)?; + activity.status = "completed".into(); + activity.exception = None; + activity.closed_at = Some(closed_at); + let mut payload = activity.payload()?; + let metadata = super::selections::activity_metadata(activity.parallel_group_path.clone())?; payload["activity_attempt_id"] = json!(attempt_id); payload["attempt_number"] = json!(DB::number(&attempt, "attempt_number")?); payload["result"] = wire(&result); @@ -4243,6 +4376,8 @@ where Err(_) => { sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?; tx.run_tags.clear(); + // The claim may have appended history the rollback took back. + tx.forget_history(); Ok(None) } } @@ -4972,7 +5107,7 @@ where // charged before the update (Workflow #930). tx.charge_bytes(super::history_meter::php_json_len(&payload).saturating_sub(recorded))?; // The only in-place history change: a cached read would miss it. - tx.scope_history = None; + tx.forget_history(); Self::query("UPDATE workflow_history_events SET payload=$1 WHERE id=$2") .bind(DB::document(&payload)) .bind(DB::string(&event, "id")?) @@ -5029,64 +5164,18 @@ where } pub(super) fn activity_payload(activity: &DB::Row) -> Result<Value> { - let mut payload = json!({"activity_execution_id": DB::string(activity,"id")?, "activity_type": DB::string(activity,"activity_type")?, - "activity_class": DB::string(activity,"activity_class")?, "sequence": DB::number(activity,"sequence")?, - "activity": {"id":DB::string(activity,"id")?,"idempotency_key":DB::string(activity,"id")?, - "sequence":DB::number(activity,"sequence")?,"type":DB::string(activity,"activity_type")?, - "class":DB::string(activity,"activity_class")?,"local_activity":false, - "activity_type": DB::string(activity,"activity_type")?, "arguments": DB::optional_string(activity,"arguments")?.map(|arguments| wire(&arguments)), - "retry_policy":DB::optional_document_row(activity,"retry_policy")?, - "status":DB::string(activity,"status")?,"attempt_count":DB::number(activity,"attempt_count")?, - "closed_at":DB::optional_instant(activity,"closed_at")?, - "attempt_id":DB::optional_string(activity,"current_attempt_id")?, - "exception":DB::optional_string(activity,"exception")?.map(|value|wire(&value)), - "payload_codec": "avro", "queue": DB::string(activity,"queue")?}}); - let metadata = super::selections::activity_metadata(DB::optional_document_row( - activity, - "parallel_group_path", - )?)?; - super::selections::merge(&mut payload, &metadata); - // `ActivitySnapshot::fromExecution` names the options' cancellation - // policy and scope when set. - if let Some(options) = DB::optional_document_row(activity, "activity_options")? { - for field in ["cancellation_policy", "cancellation_scope_id"] { - if let Some(value) = options[field].as_str() { - payload["activity"][field] = json!(value); - } - } - // `ActivitySnapshot::fromExecution`: an abandoned activity names - // the lifetime it keeps after the run stops waiting. - if options["cancellation_policy"] == "abandon" - && let Some(deadline) = DB::optional_instant(activity, "schedule_to_close_deadline_at")? - { - payload["activity"]["schedule_to_close_deadline_at"] = json!(deadline.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()); - } - } - // Released history readers retain the first claim and heartbeat across - // a retry even while the next attempt has not produced new progress. - for field in ["started_at", "last_heartbeat_at"] { - if let Some(value) = DB::optional_instant(activity, field)? { - payload["activity"][field] = json!(value); - } - } - if payload["activity"]["arguments"].is_null() { - payload["activity"].as_object_mut().unwrap().remove("arguments"); - } - if payload["activity"]["retry_policy"].is_null() { - // A legacy absent snapshot stays absent in its durable history. - // The activity claim may still advertise a nullable policy. - payload["activity"] - .as_object_mut() - .unwrap() - .remove("retry_policy"); - } - Ok(payload) + ActivityRecord::read::<DB>(activity)?.payload() } fn event(row: DB::Row) -> Result<Value> { - let event_type = DB::string(&row, "event_type")?; - let payload = - super::history_presentation::payload(DB::document_row(&row, "payload")?, &event_type); + let payload = DB::document_row(&row, "payload")?; + Self::event_with(&row, payload) + } + + /// `event` with the row's stored payload, parsed. + fn event_with(row: &DB::Row, payload: Value) -> Result<Value> { + let event_type = DB::string(row, "event_type")?; + let payload = super::history_presentation::payload(payload, &event_type); Ok( json!({"id": DB::string(&row,"id")?, "sequence": DB::number(&row,"sequence")?, "event_type": DB::string(&row,"event_type")?, "payload": payload, @@ -5147,12 +5236,18 @@ where Ok(()) } - /// A run's whole history, whose last sequence is `last_sequence`. - async fn whole_history(connection: &mut DB::Connection, run_id: &str, last_sequence: i64) -> Result<Vec<Value>> { + /// A run's whole history, whose last sequence is `last_sequence`, and + /// each event recorded in `fold`. + pub(super) async fn whole_history( + connection: &mut DB::Connection, + run_id: &str, + last_sequence: i64, + mut fold: Option<&mut SummaryFold>, + ) -> Result<Vec<Value>> { let mut events = Vec::new(); let mut after = 0; loop { - let (page, next) = Self::history_page_known(&mut *connection, run_id, after, 1000, Some(last_sequence)).await?; + let (page, next) = Self::history_page_known(&mut *connection, run_id, after, 1000, Some(last_sequence), fold.as_deref_mut()).await?; if let Some(last) = page.last() { after = last["sequence"].as_i64().unwrap_or(after); } @@ -5169,18 +5264,20 @@ where after_sequence: i64, page_size: i64, ) -> Result<(Vec<Value>, Option<String>)> { - Self::history_page_known(connection, run_id, after_sequence, page_size, None).await + Self::history_page_known(connection, run_id, after_sequence, page_size, None, None).await } /// `history_page_connection` for a caller that has read the run's /// `last_history_sequence` in this transaction, or `None` to read it - /// after the page. + /// after the page. `fold` records each event, from the payload text the + /// summary scan reads. async fn history_page_known( connection: &mut DB::Connection, run_id: &str, after_sequence: i64, page_size: i64, last_sequence: Option<i64>, + fold: Option<&mut SummaryFold>, ) -> Result<(Vec<Value>, Option<String>)> { use base64::{Engine, engine::general_purpose::STANDARD}; // Bound serialized history bytes before hydrating JSON. SQLite executes @@ -5212,10 +5309,19 @@ where )); } let next = (last < total).then(|| STANDARD.encode(last.to_string())); - Ok(( - rows.into_iter().map(Self::event).collect::<Result<_>>()?, - next, - )) + let events = match fold { + None => rows.into_iter().map(Self::event).collect::<Result<_>>()?, + Some(fold) => rows + .iter() + .map(|row| { + let raw = DB::document_text(row, "payload")?; + let payload = serde_json::from_str(&raw)?; + fold.record(&DB::string(row, "id")?, DB::number(row, "sequence")?, &DB::string(row, "event_type")?, &raw, &payload); + Self::event_with(row, payload) + }) + .collect::<Result<_>>()?, + }; + Ok((events, next)) } } @@ -5519,6 +5625,10 @@ pub(super) struct Apply<'a, R> { /// The first child this batch started with `await: "started"`: the /// parent's start turn (Server #966). pub(super) start_turn: &'a std::sync::Mutex<Option<Value>>, + /// The run's pending and running activities, as read before the batch's + /// first command; that command takes it, so no later one reads a count + /// a command may have changed. + pub(super) open_activities: &'a std::sync::Mutex<Option<i64>>, } impl<R> Clone for Apply<'_, R> { @@ -5754,3 +5864,110 @@ pub(super) struct StartMetadata<'a> { pub(super) memo: Option<&'a super::memos::PreparedMemo>, pub(super) search_attributes: Option<&'a super::search_attribute_upserts::StartSearchAttributes>, } + +/// The activity execution columns its history payload and claim read, as a +/// row holds them, or as a statement of this transaction has just written +/// them, so the transaction need not read its own write back. +pub(super) struct ActivityRecord { + pub(super) id: String, + pub(super) activity_type: String, + pub(super) activity_class: String, + pub(super) sequence: i64, + pub(super) arguments: Option<String>, + pub(super) retry_policy: Option<Value>, + pub(super) status: String, + pub(super) attempt_count: i64, + pub(super) closed_at: Option<DateTime<Utc>>, + pub(super) current_attempt_id: Option<String>, + pub(super) exception: Option<String>, + pub(super) queue: String, + pub(super) parallel_group_path: Option<Value>, + pub(super) activity_options: Option<Value>, + pub(super) started_at: Option<DateTime<Utc>>, + pub(super) last_heartbeat_at: Option<DateTime<Utc>>, + /// The schedule-to-start, start-to-close, schedule-to-close and + /// heartbeat deadlines (`activity_deadlines::deadlines_of`). + pub(super) deadlines: [Option<DateTime<Utc>>; 4], +} + +impl ActivityRecord { + pub(super) fn read<DB: Backend>(activity: &DB::Row) -> Result<Self> { + Ok(Self { + id: DB::string(activity, "id")?, + activity_type: DB::string(activity, "activity_type")?, + activity_class: DB::string(activity, "activity_class")?, + sequence: DB::number(activity, "sequence")?, + arguments: DB::optional_string(activity, "arguments")?, + retry_policy: DB::optional_document_row(activity, "retry_policy")?, + status: DB::string(activity, "status")?, + attempt_count: DB::number(activity, "attempt_count")?, + closed_at: DB::optional_instant(activity, "closed_at")?, + current_attempt_id: DB::optional_string(activity, "current_attempt_id")?, + exception: DB::optional_string(activity, "exception")?, + queue: DB::string(activity, "queue")?, + parallel_group_path: DB::optional_document_row(activity, "parallel_group_path")?, + activity_options: DB::optional_document_row(activity, "activity_options")?, + started_at: DB::optional_instant(activity, "started_at")?, + last_heartbeat_at: DB::optional_instant(activity, "last_heartbeat_at")?, + deadlines: [ + DB::optional_instant(activity, "schedule_deadline_at")?, + DB::optional_instant(activity, "close_deadline_at")?, + DB::optional_instant(activity, "schedule_to_close_deadline_at")?, + DB::optional_instant(activity, "heartbeat_deadline_at")?, + ], + }) + } + + /// The activity's history payload (`Store::activity_payload`). + pub(super) fn payload(&self) -> Result<Value> { + let mut payload = json!({"activity_execution_id": self.id, "activity_type": self.activity_type, + "activity_class": self.activity_class, "sequence": self.sequence, + "activity": {"id":self.id,"idempotency_key":self.id, + "sequence":self.sequence,"type":self.activity_type, + "class":self.activity_class,"local_activity":false, + "activity_type": self.activity_type, "arguments": self.arguments.as_deref().map(wire), + "retry_policy":self.retry_policy, + "status":self.status,"attempt_count":self.attempt_count, + "closed_at":self.closed_at, + "attempt_id":self.current_attempt_id, + "exception":self.exception.as_deref().map(wire), + "payload_codec": "avro", "queue": self.queue}}); + let metadata = super::selections::activity_metadata(self.parallel_group_path.clone())?; + super::selections::merge(&mut payload, &metadata); + // `ActivitySnapshot::fromExecution` names the options' cancellation + // policy and scope when set. + if let Some(options) = &self.activity_options { + for field in ["cancellation_policy", "cancellation_scope_id"] { + if let Some(value) = options[field].as_str() { + payload["activity"][field] = json!(value); + } + } + // `ActivitySnapshot::fromExecution`: an abandoned activity names + // the lifetime it keeps after the run stops waiting. + if options["cancellation_policy"] == "abandon" + && let Some(deadline) = self.deadlines[2] + { + payload["activity"]["schedule_to_close_deadline_at"] = json!(deadline.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()); + } + } + // Released history readers retain the first claim and heartbeat across + // a retry even while the next attempt has not produced new progress. + for (field, value) in [("started_at", self.started_at), ("last_heartbeat_at", self.last_heartbeat_at)] { + if let Some(value) = value { + payload["activity"][field] = json!(value); + } + } + if payload["activity"]["arguments"].is_null() { + payload["activity"].as_object_mut().unwrap().remove("arguments"); + } + if payload["activity"]["retry_policy"].is_null() { + // A legacy absent snapshot stays absent in its durable history. + // The activity claim may still advertise a nullable policy. + payload["activity"] + .as_object_mut() + .unwrap() + .remove("retry_policy"); + } + Ok(payload) + } +} diff --git a/rust/src/runtime/terminal_summary.rs b/rust/src/runtime/terminal_summary.rs index ea9d2a31..6a3032e0 100644 --- a/rust/src/runtime/terminal_summary.rs +++ b/rust/src/runtime/terminal_summary.rs @@ -50,10 +50,7 @@ where tx: &mut Transaction<'_, DB>, run_id: &str, ) -> Result<()> { - let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1") - .bind(run_id) - .fetch_one(&mut **tx) - .await?; + let (run, census) = Self::census_run(tx, run_id).await?; let status = DB::string(&run, "status")?; let closed = terminal(&status); // Retention and v1 projections need their separately qualified @@ -85,7 +82,6 @@ where updates, .. } = history; - let census = Self::projection_census(tx, run_id).await?; let active = if closed { None } else { @@ -199,18 +195,24 @@ where } } // Missing/incomplete summaries use the same bounded canonical scan. - // This calculation does not persist or repair any projection. - let history = Self::summary_history(tx, run_id, true, last) + // This calculation does not persist or repair any projection. It + // scans as an open run's projection does, and the transaction keeps + // the scan, so the run's projection at commit continues it instead + // of reading the history again. + let history = Self::scan_summary_history(tx, run_id, false, last) .await? + .filter(|history| history.count == last && history.count != 0) .ok_or_else(|| { super::refuse( axum::http::StatusCode::CONFLICT, "history_budget_unavailable", ) })?; - Ok(tx + let budget = tx .history_budget - .summarize(history.count, history.bytes, history.fan_out)) + .summarize(history.count, history.bytes, history.fan_out); + tx.summary_scan = Some(SummaryScan { run_id: run_id.to_owned(), open: true, history }); + Ok(budget) } async fn summary_history( @@ -219,9 +221,45 @@ where closed: bool, last: i64, ) -> Result<Option<SummaryHistory>> { - let mut history = SummaryHistory { - active_history_supported: true, - ..SummaryHistory::default() + let Some(mut history) = Self::scan_summary_history(tx, run_id, closed, last).await? else { + return Ok(None); + }; + history.active_history_supported &= history.selections.complete(); + history.active_history_supported &= history.barriers.complete(); + Ok((history.count == last && history.count != 0).then_some(history)) + } + + /// `summary_history`'s pass over the events, before it judges the + /// open selections and barriers: from the scan this transaction kept + /// for the run, when there is one that covers what `closed` records, + /// or else from the first event. `None` when a sequence is missing. + async fn scan_summary_history( + tx: &mut Transaction<'_, DB>, + run_id: &str, + closed: bool, + last: i64, + ) -> Result<Option<SummaryHistory>> { + let kept = match tx.summary_scan.take() { + Some(scan) if scan.run_id != run_id => { + tx.summary_scan = Some(scan); + None + } + scan => scan.filter(|scan| (closed || scan.open) && scan.history.count <= last), + }; + // A kept scan read to a short page, so the history ended there. A + // later event takes a sequence past it, so a kept scan that has + // reached `last` has read every event. + if let Some(scan) = kept.as_ref() + && scan.history.count == last + { + return Ok(kept.map(|scan| scan.history)); + } + let mut history = match kept { + Some(scan) => scan.history, + None => SummaryHistory { + active_history_supported: true, + ..SummaryHistory::default() + }, }; loop { // Bound hydrated history to sixteen events. The transaction already @@ -248,9 +286,56 @@ where break; } } - history.active_history_supported &= history.selections.complete(); - history.active_history_supported &= history.barriers.complete(); - Ok((history.count == last && history.count != 0).then_some(history)) + Ok(Some(history)) + } +} + +/// A summary scan of one run's history that a transaction keeps for the +/// run's projection. It lives in the transaction, which holds the transition +/// lock, so the history only grows by appended sequences; the in-place +/// update and the deletes forget it (`Transaction::forget_history`). `open` +/// is whether it recorded what an open run's projection records. +pub(super) struct SummaryScan { + run_id: String, + open: bool, + history: SummaryHistory, +} + +/// A summary scan built from history rows the transaction read from the +/// first event for another purpose (a workflow task's history), so the +/// run's budget and projection do not read them again. +pub(super) struct SummaryFold(Option<SummaryHistory>); + +impl Default for SummaryFold { + fn default() -> Self { + Self(Some(SummaryHistory { + active_history_supported: true, + ..SummaryHistory::default() + })) + } +} + +impl SummaryFold { + /// Records the next event as an open run's scan does. An event out of + /// sequence, or one the scan refuses, abandons the fold, so the scan + /// reads and judges the history itself. + pub(super) fn record(&mut self, id: &str, sequence: i64, kind: &str, raw: &str, payload: &Value) { + let Some(history) = self.0.as_mut() else { + return; + }; + if sequence == history.count + 1 && history.record_parsed(id, kind, raw, payload, false).is_ok() { + history.count = sequence; + } else { + self.0 = None; + } + } + + /// Keeps the fold as the transaction's scan when it recorded the run's + /// whole history, through `last`. + pub(super) fn keep<DB: Backend>(self, tx: &mut Transaction<'_, DB>, run_id: &str, last: i64) { + if let Some(history) = self.0.filter(|history| history.count == last) { + tx.summary_scan = Some(SummaryScan { run_id: run_id.to_owned(), open: true, history }); + } } } @@ -274,11 +359,15 @@ struct SummaryHistory { impl SummaryHistory { fn record(&mut self, id: &str, kind: &str, raw: &str, closed: bool) -> Result<()> { - let payload: Value = serde_json::from_str(raw)?; - self.bytes += history_budget::parsed_event_size(kind, raw, &payload)? as i64; + self.record_parsed(id, kind, raw, &serde_json::from_str(raw)?, closed) + } + + /// `record` for an event whose payload text the caller has parsed. + fn record_parsed(&mut self, id: &str, kind: &str, raw: &str, payload: &Value, closed: bool) -> Result<()> { + self.bytes += history_budget::parsed_event_size(kind, raw, payload)? as i64; if !closed { - self.active_history_supported &= self.barriers.apply(kind, &payload); - self.active_history_supported &= self.selections.apply(id, kind, &payload); + self.active_history_supported &= self.barriers.apply(kind, payload); + self.active_history_supported &= self.selections.apply(id, kind, payload); let supported = if kind == "SelectionResolved" { true } else if matches!(kind, "TimerScheduled" | "TimerFired" | "TimerCancelled") { @@ -286,26 +375,26 @@ impl SummaryHistory { payload["timer_kind"].as_str(), Some("signal_timeout" | "condition_timeout") ) { - self.external_waits.apply(kind, &payload) + self.external_waits.apply(kind, payload) } else { - self.timers.apply(kind, &payload) + self.timers.apply(kind, payload) } } else if kind.starts_with("Activity") { - self.activities.apply(kind, &payload) + self.activities.apply(kind, payload) } else if kind.starts_with("Child") { - self.children.apply(kind, &payload) + self.children.apply(kind, payload) } else if kind.starts_with("Update") { - self.updates.apply(kind, &payload) + self.updates.apply(kind, payload) } else { - self.external_waits.apply(kind, &payload) + self.external_waits.apply(kind, payload) }; self.active_history_supported &= supported; } if kind == "WorkflowStarted" && self.started.is_none() { self.started = Some(payload.clone()); } - self.record_fan_out(&payload); - self.record_failure(kind, &payload); + self.record_fan_out(payload); + self.record_failure(kind, payload); Ok(()) } diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs index 5ad98151..8d7f1466 100644 --- a/rust/src/runtime/transaction.rs +++ b/rust/src/runtime/transaction.rs @@ -23,11 +23,18 @@ pub(super) struct Transaction<'a, DB: Backend> { /// has committed or rolled back (`Store::begin`). pub(super) queued: Option<tokio::sync::OwnedMutexGuard<()>>, summaries: BTreeSet<String>, + /// The runs whose waits this transaction opened or reopened. Ordinals + /// rank a run's waits by when each opened, and only all of a run's waits + /// are ever deleted, so no other change moves one (`Store::order_waits`). + wait_order: BTreeSet<String>, pub(super) history_budget: super::history_budget::Config, /// A workflow task completion's history meter (Workflow #930). pub(super) meter: Option<super::history_meter::Meter>, /// The run history the cancellation scope engine has read (`scope_events`). pub(super) scope_history: Option<super::cancellation_scopes::HistoryCache>, + /// The summary scan of a run's history this transaction has kept + /// (`Store::workflow_history_budget`), for the run's projection. + pub(super) summary_scan: Option<super::terminal_summary::SummaryScan>, /// Times the transition lock while it is held (`DW_BENCH_TRACE`). pub(super) lock_trace: Option<super::bench_trace::LockTrace>, /// The binding key of a workflow poll request this transaction has @@ -38,6 +45,9 @@ pub(super) struct Transaction<'a, DB: Backend> { pub(super) run_tags: HashMap<String, RunTags>, /// Local activity rows not yet written (`Store::write_local_activities`). pub(super) local_writes: super::local_activities::Writes, + /// The activities this transaction scheduled and has not yet started an + /// attempt of: their ids are new, so they have no attempt to expire. + pub(super) unattempted_activities: BTreeSet<String>, } impl<'a, DB: Backend> Transaction<'a, DB> { @@ -49,13 +59,16 @@ impl<'a, DB: Backend> Transaction<'a, DB> { inner, queued: None, summaries: BTreeSet::new(), + wait_order: BTreeSet::new(), history_budget, meter: None, scope_history: None, + summary_scan: None, lock_trace: None, replayed_poll: None, run_tags: HashMap::new(), local_writes: Default::default(), + unattempted_activities: BTreeSet::new(), } } @@ -63,6 +76,17 @@ impl<'a, DB: Backend> Transaction<'a, DB> { self.summaries.insert(run_id.to_owned()); } + pub(super) fn wait_order_changed(&mut self, run_id: &str) { + self.wait_order.insert(run_id.to_owned()); + } + + /// Forget every history read this transaction keeps: its rows were + /// changed in place or deleted, so a later read starts over. + pub(super) fn forget_history(&mut self) { + self.scope_history = None; + self.summary_scan = None; + } + /// Rows the next statement will write, charged to a metered completion. pub(super) fn charge_rows(&mut self, rows: usize) -> Result<()> { match self.meter.as_mut() { @@ -129,7 +153,9 @@ where let written = std::time::Instant::now(); super::bench_trace::phase("local writes", written - started); for run_id in std::mem::take(&mut self.summaries) { - Store::<DB>::order_waits(&mut self, &run_id).await?; + if self.wait_order.contains(&run_id) { + Store::<DB>::order_waits(&mut self, &run_id).await?; + } Store::<DB>::project_run_summary(&mut self, &run_id).await?; } let projected = std::time::Instant::now(); diff --git a/rust/src/runtime/wait_projection.rs b/rust/src/runtime/wait_projection.rs index 0627a4c1..fb617ef1 100644 --- a/rust/src/runtime/wait_projection.rs +++ b/rust/src/runtime/wait_projection.rs @@ -20,7 +20,8 @@ where { /// Store each wait's chronological ordinal over all of the run's retained /// waits, as Workflow #941's projector does, writing only waits that moved. - /// Resolving a wait moves nothing, so most transitions write no positions. + /// Resolving a wait moves nothing, so a transition that opened no wait + /// does not rank them (`Transaction::wait_order_changed`). pub(super) async fn order_waits(tx: &mut Transaction<'_, DB>, run: &str) -> Result<()> { let mut moved = std::collections::HashSet::new(); loop { @@ -74,6 +75,7 @@ where .bind(projection_id(run, &wait)).bind(wait).bind(sequence - 1).bind(format!("Waiting for update {name}.")) .bind(DB::bind_time(at)).bind(name).bind(update).bind(super::text(payload, "workflow_command_id")?) .bind(sequence).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(run).execute(&mut **tx).await?; + tx.wait_order_changed(run); } else { let failed = payload["failure_id"] .as_str() @@ -122,6 +124,7 @@ where .bind(DB::bind_time(at)).bind(super::text(payload,"child_workflow_instance_id")?).bind(label) .bind(super::text(payload,"child_workflow_run_id")?).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(run_id) .execute(&mut **tx).await?; + tx.wait_order_changed(run_id); } else { // `ChildRunHistory::waitSnapshotForSequence`: a cancelled or // terminated child ends the wait as cancelled. @@ -147,21 +150,47 @@ where activity_id: &str, task_id: &str, at: DateTime<Utc>, + ) -> Result<()> { + let activity = Self::query("SELECT a.id,a.workflow_run_id,a.sequence,a.activity_type,r.workflow_instance_id FROM activity_executions a JOIN workflow_runs r ON r.id=a.workflow_run_id WHERE a.id=$1 AND a.workflow_run_id=$2") + .bind(activity_id).bind(run_id).fetch_one(&mut **tx).await?; + let activity_type = DB::string(&activity, "activity_type")?; + Self::open_activity_wait( + tx, + &DB::string(&activity, "workflow_instance_id")?, + &DB::string(&activity, "workflow_run_id")?, + &DB::string(&activity, "id")?, + DB::number(&activity, "sequence")?, + &activity_type, + task_id, + at, + ) + .await + } + + /// `create_activity_wait` for an activity, of a run in the given + /// instance, whose row the transaction has already read or written. + #[allow(clippy::too_many_arguments)] + pub(super) async fn open_activity_wait( + tx: &mut Transaction<'_, DB>, + instance_id: &str, + run_id: &str, + activity_id: &str, + sequence: i64, + activity_type: &str, + task_id: &str, + at: DateTime<Utc>, ) -> Result<()> { // Only the authored ActivityScheduled transition creates this row. // Reads and later events never repair an absent projection implicitly. - let activity = Self::query("SELECT a.sequence,a.activity_type,r.workflow_instance_id FROM activity_executions a JOIN workflow_runs r ON r.id=a.workflow_run_id WHERE a.id=$1 AND a.workflow_run_id=$2") - .bind(activity_id).bind(run_id).fetch_one(&mut **tx).await?; let wait_id = format!("activity:{activity_id}"); let projection_id = projection_id(run_id, &wait_id); - let sequence = DB::number(&activity, "sequence")?; - let activity_type = DB::string(&activity, "activity_type")?; Self::query("INSERT INTO workflow_run_waits(id,workflow_run_id,workflow_instance_id,wait_id,position,kind,sequence,status,source_status,summary,opened_at,target_type,task_backed,external_only,resume_source_kind,resume_source_id,task_id,task_type,task_status,history_authority,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,'activity',$6,'open','pending',$7,$8,$9,TRUE,FALSE,'activity_execution',$10,$11,'activity','ready','typed_history',$12,$13)") - .bind(projection_id).bind(run_id).bind(DB::string(&activity, "workflow_instance_id")?) + .bind(projection_id).bind(run_id).bind(instance_id) .bind(wait_id).bind(sequence - 1).bind(sequence) .bind(format!("Waiting for activity {activity_type}.")) .bind(DB::bind_time(at)).bind(activity_type).bind(activity_id).bind(task_id) .bind(DB::bind_time(at)).bind(DB::bind_time(at)).execute(&mut **tx).await?; + tx.wait_order_changed(run_id); Ok(()) } @@ -197,6 +226,7 @@ where } else if event_type == "ActivityStarted" { Self::query("UPDATE workflow_run_waits SET source_status='running',task_status='leased',opened_at=$1,updated_at=$2 WHERE workflow_run_id=$3 AND wait_id=$4 AND status='open'") .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(run_id).bind(wait_id).execute(&mut **tx).await?; + tx.wait_order_changed(run_id); } else { let activity_type = super::text(payload, "activity_type")?; let (status, source_status, summary) = match event_type { @@ -242,6 +272,7 @@ where .bind(DB::bind_time(at)).bind(DB::bind_time(fire_at)).bind(timer_id).bind(task_id) .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(run_id) .execute(&mut **tx).await?; + tx.wait_order_changed(run_id); Ok(()) } @@ -299,6 +330,7 @@ where .bind(summary) .bind(DB::bind_time(at)).bind(key).bind(DB::bind_time(at)).bind(DB::bind_time(at)) .bind(run_id).execute(&mut **tx).await?; + tx.wait_order_changed(run_id); } else if event_type == "ConditionWaitSatisfied" { // Receipt alone is not satisfaction. Only the worker's committed // application event resolves this occurrence, without read repair. @@ -362,6 +394,7 @@ where .bind(summary) .bind(DB::bind_time(at)).bind(name).bind(DB::bind_time(at)).bind(DB::bind_time(at)) .bind(run_id).execute(&mut **tx).await?; + tx.wait_order_changed(run_id); } else if payload["timer_kind"] == "signal_timeout" && matches!( event_type, diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs index 7e1b538a..17891d78 100644 --- a/rust/src/runtime/workflow_poll_requests.rs +++ b/rust/src/runtime/workflow_poll_requests.rs @@ -18,12 +18,16 @@ use super::transaction::Transaction; use super::{Result, RuntimeError, store::now}; use axum::http::StatusCode; use chrono::{DateTime, Utc}; -use serde_json::{Value, json}; +use serde_json::{Map, Value, json}; use sha2::{Digest, Sha256}; // Distinct from the query poll prefix, so each sweep sees only its own rows. const PREFIX: &str = "dw-native-workflow-poll:"; +// A remembered workflow lease keeps where its history ended in place of the +// events (`remembered`), and a replay reads them again (`with_history`). +const HISTORY: &str = "remembered_history"; + // The task row a cached workflow lease is delivered from. const WORKFLOW_LEASE: &str = "SELECT status,queue,lease_owner,lease_expires_at,attempt_count FROM workflow_tasks WHERE id=$1 AND task_type='workflow'"; @@ -93,6 +97,39 @@ fn task_kind(response: &Value) -> &str { response["task"]["task_kind"].as_str().unwrap_or("workflow") } +/// An answer as it is remembered. A workflow lease's history events are +/// the run's events through the last one delivered, which never change +/// while the lease can be delivered, so it keeps that sequence and the +/// signal sequences the events showed instead of the events. +fn remembered(response: &Value) -> Value { + let (Some(answer), Some(events)) = (response.as_object(), response["task"]["history_events"].as_array()) else { + return response.clone(); + }; + let through = events.last().and_then(|event| event["sequence"].as_i64()).unwrap_or(0); + let signal_sequences: Map<String, Value> = events + .iter() + .filter(|event| event["event_type"] == "SignalReceived") + .filter_map(|event| { + let sequence = event["payload"].get("workflow_sequence").filter(|sequence| !sequence.is_null())?; + Some((event["payload"]["signal_id"].as_str()?.to_owned(), sequence.clone())) + }) + .collect(); + let mut kept = Map::new(); + for (key, value) in answer { + let value = match value.as_object().filter(|_| key == "task") { + Some(task) => { + let task = task.iter().map(|(field, value)| (field.clone(), if field == "history_events" { Value::Null } else { value.clone() })); + Value::Object(task.collect()) + } + None => value.clone(), + }; + kept.insert(key.clone(), value); + } + let run_id = response["task"]["run_id"].clone(); + kept.insert(HISTORY.into(), json!({"run_id": run_id, "through": through, "signal_sequences": signal_sequences})); + Value::Object(kept) +} + /// The 409 for an ID already bound to another task-kind set. fn kinds_conflict(request: &WorkflowPollRequest, bound: Value) -> RuntimeError { RuntimeError::Protocol { @@ -137,7 +174,9 @@ store_methods! { .collect(); format!("{PREFIX}{digest}") }; - let keys = WorkflowPollKeys { binding: key("binding"), result: key("result") }; + // A remembered lease is not the answer it delivered, so it has its + // own key: a server that keeps whole answers never reads one. + let keys = WorkflowPollKeys { binding: key("binding"), result: key("remembered-result") }; Ok(WorkflowPollRequest { keys, pending_ttl: pending_ttl(default_wait), kinds: normalized(kinds), id }) } @@ -206,9 +245,10 @@ store_methods! { /// The cached answer if its task can still be delivered; a stale answer /// is deleted. A workflow lease shows the task's current lease, as PHP - /// rebuilds it; a validation lease replays as it was delivered. An - /// answer of a kind this request did not ask for is not replayed. - /// `result` is the answer `bind_workflow_poll` read. + /// rebuilds it, and the history it delivered, read again; a validation + /// lease replays as it was delivered. An answer of a kind this request + /// did not ask for is not replayed. `result` is the answer + /// `bind_workflow_poll` read. pub(super) async fn deliverable_workflow_poll( &self, tx: &mut Transaction<'_, DB>, @@ -233,6 +273,7 @@ store_methods! { Self::workflow_lease_deliverable(tx, &mut result["task"], worker_id, queue).await? }; if deliverable { + Self::with_history(&mut **tx, &mut result).await?; return Ok(Some(result)); } Self::query(DB::QUERY_CACHE_DELETE).bind(&request.keys.result).execute(&mut **tx).await?; @@ -280,6 +321,30 @@ store_methods! { Ok(false) } + /// A remembered lease's history events, read again: the run's events + /// through the sequence the delivered history ended at, showing the + /// signal sequences they showed then (`remembered`). + async fn with_history(connection: &mut DB::Connection, answer: &mut Value) -> Result<()> { + let Some(kept) = answer.as_object_mut().and_then(|answer| answer.shift_remove(HISTORY)) else { + return Ok(()); + }; + let (Some(run_id), Some(through)) = (kept["run_id"].as_str(), kept["through"].as_i64()) else { + return Err(RuntimeError::Refused { status: StatusCode::INTERNAL_SERVER_ERROR, reason: "remembered_workflow_lease_unreadable" }); + }; + let mut events = if through > 0 { Self::whole_history(connection, run_id, through, None).await? } else { Vec::new() }; + events.retain(|event| event["sequence"].as_i64().is_some_and(|sequence| sequence <= through)); + for event in events.iter_mut().filter(|event| event["event_type"] == "SignalReceived") { + let sequence = event["payload"]["signal_id"].as_str().and_then(|id| kept["signal_sequences"].get(id)).cloned(); + if let (Some(sequence), Some(payload)) = (sequence, event["payload"].as_object_mut()) + && payload.get("workflow_sequence").is_none_or(Value::is_null) + { + payload.insert("workflow_sequence".into(), sequence); + } + } + answer["task"]["history_events"] = Value::Array(events); + Ok(()) + } + /// The broker's `cachedTaskStillDeliverable`: the validation task is in /// this namespace and queue, leased to this worker with a live lease, at /// the delivered attempt. @@ -307,13 +372,14 @@ store_methods! { /// Remember a poll's answer under its request ID, and rewrite the /// binding with PHP's result expiry. A workflow lease's answer is kept /// while its task can be delivered (the row decides), as PHP's lease - /// binding replays for the whole live lease. + /// binding replays for the whole live lease. A workflow lease is kept + /// without its history events (`remembered`). pub(super) async fn remember_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest, response: &Value) -> Result<()> { let ttl = result_ttl(response, request.pending_ttl); let kept = if response["task"].is_object() && task_kind(response) == "workflow" { 3600 } else { ttl }; Self::query(DB::QUERY_CACHE_PUT_PAIR) .bind(&request.keys.binding).bind(json!({"task_kinds": request.kinds}).to_string()).bind(epoch() + ttl) - .bind(&request.keys.result).bind(response.to_string()).bind(epoch() + kept) + .bind(&request.keys.result).bind(remembered(response).to_string()).bind(epoch() + kept) .execute(&mut **tx) .await?; Ok(()) @@ -348,7 +414,11 @@ store_methods! { return Ok(None); }; let row = Self::query(WORKFLOW_LEASE).bind(&task_id).fetch_optional(&self.pool).await?; - Ok(Self::lease_row_deliverable(row, &mut result["task"], worker_id, queue)?.then_some(result)) + if !Self::lease_row_deliverable(row, &mut result["task"], worker_id, queue)? { + return Ok(None); + } + Self::with_history(&mut *self.pool.acquire().await?, &mut result).await?; + Ok(Some(result)) } /// A cache row read without the lock, if it has not expired. diff --git a/rust/tests/execution/activity_results.rs b/rust/tests/execution/activity_results.rs index 86a50a27..096ff29b 100644 --- a/rust/tests/execution/activity_results.rs +++ b/rust/tests/execution/activity_results.rs @@ -404,3 +404,115 @@ async fn activity_poll_answers_carry_the_server_clock() { assert_eq!(empty.1["poll_status"], "empty", "{}", empty.1); assert!(micros(&empty.1["server_time"]), "{}", empty.1); } + +/// An activity's row, each column as its text or null. +async fn activity_row(database: &TestDatabase, id: &str) -> Value { + let cast = if matches!(database, TestDatabase::MySql { .. }) { "CHAR" } else { "TEXT" }; + let query = [ + "status", "attempt_count", "current_attempt_id", "queue", "retry_policy", "started_at", "closed_at", + "last_heartbeat_at", "schedule_deadline_at", "close_deadline_at", "schedule_to_close_deadline_at", + "heartbeat_deadline_at", + ] + .map(|column| format!("SELECT '{column}',CAST({column} AS {cast}) FROM activity_executions WHERE id='{id}'")) + .join(" UNION ALL "); + macro_rules! read { + ($pool:expr) => {{ + let pool = $pool; + let rows: Vec<(String, Option<String>)> = + sqlx::query_as(sqlx::AssertSqlSafe(query)).fetch_all(&pool).await.unwrap(); + pool.close().await; + rows + }}; + } + let rows = match database { + TestDatabase::Sqlite(dir) => read!( + sqlx::SqlitePool::connect(&format!("sqlite://{}", dir.path().join("runtime.sqlite").display())).await.unwrap() + ), + TestDatabase::Postgres { options, .. } => { + read!(sqlx::postgres::PgPoolOptions::new().connect_with(options.as_ref().clone()).await.unwrap()) + } + TestDatabase::MySql { options, .. } => { + read!(sqlx::mysql::MySqlPoolOptions::new().connect_with(options.as_ref().clone()).await.unwrap()) + } + }; + Value::Object(rows.into_iter().map(|(column, value)| (column, json!(value))).collect()) +} + +/// A time as a history snapshot or a row's text holds it. +fn snapshot_time(value: &Value) -> Option<chrono::NaiveDateTime> { + let text = value.as_str()?; + Some( + chrono::DateTime::parse_from_rfc3339(text) + .map(|at| at.naive_utc()) + .or_else(|_| chrono::NaiveDateTime::parse_from_str(text, "%Y-%m-%d %H:%M:%S%.f")) + .unwrap(), + ) +} + +/// An activity's history snapshot agrees with its row. +fn assert_snapshot_is_row(snapshot: &Value, row: &Value) { + assert_eq!(snapshot["status"], row["status"], "{snapshot} {row}"); + assert_eq!(snapshot["attempt_count"].to_string(), row["attempt_count"].as_str().unwrap(), "{snapshot} {row}"); + assert_eq!(snapshot["attempt_id"], row["current_attempt_id"], "{snapshot} {row}"); + assert_eq!(snapshot["queue"], row["queue"], "{snapshot} {row}"); + let policy: Value = serde_json::from_str(row["retry_policy"].as_str().unwrap()).unwrap(); + assert_eq!(snapshot["retry_policy"], policy, "{snapshot} {row}"); + for field in ["started_at", "closed_at", "last_heartbeat_at"] { + assert_eq!(snapshot_time(&snapshot[field]), snapshot_time(&row[field]), "{field}: {snapshot} {row}"); + } + assert_eq!( + snapshot_time(&snapshot["schedule_to_close_deadline_at"]),rmcdaniel commentedon Nov 6, 2026
Sanitized external reference for topic 23:
0005-Cut-more-statements-from-each-workflow-and-activity-.patch, part 3/3. Complete sanitized diff: 103,089 bytes / SHA-2563095223e9455381ff492a08b7daf5651ce94aa138bff00cecaf3bc2bdfdc2306. Ordered parts preserve the complete diff bytes; none was applied or executed. The patch is a cumulative older-source draft, not maintained implementation or qualification. Mail headers/author/date/message/application instructions were removed.The supplied 0005 stage includes the known stale wait-ranking shortcut. It is not an acceptable production candidate by itself; retain and independently qualify correction0006 and the negative controls.
rmcdaniel commentedon Nov 6, 2026
Sanitized external reference for topic 23:
0006-Rank-a-run-s-waits-on-every-transition-again-fix-the.patch, part 1/1. Complete sanitized diff: 8,143 bytes / SHA-2564d1d1bc2e7e150495a6109f8c4ee0b2d13e65da67a7c8add9c0ef4c46f8fc090. Ordered parts preserve the complete diff bytes; none was applied or executed. The patch is a cumulative older-source draft, not maintained implementation or qualification. Mail headers/author/date/message/application instructions were removed.The supplied 0005 stage includes the known stale wait-ranking shortcut. It is not an acceptable production candidate by itself; retain and independently qualify correction0006 and the negative controls.
rmcdaniel commentedon Nov 6, 2026
Intake is filed as P0, owner server. Shared index, allocation and static review findings.
Complete sanitized diff reference(s): 0005 part 1/3, 0005 part 2/3, 0005 part 3/3, 0006 part 1/1.
Shared Lockbox bundle:
lab4/general/dw-performance-intake/2026-11-06/sanitized-tricks-20-25.zip(101,306 bytes; SHA-2563e5b593ed89536451be1c8d74abf2ee6e4782cd3f7a75fe5fc9a645d648a52da). Its roundtrip and member hashes are verified. The supplied patches and benchmark results remain external reference data; no patch was applied or executed during intake. Existing issue acceptance and subsequent owner review remain in force.rmcdaniel commentedon Nov 7, 2026
Security static reconciliation for the sanitized
0006reference, reviewed as data only:The activity-snapshot test change is a legitimate correction, not a maintained-policy weakening.
c3bbfa5fexplicitly refusesschedule_activity.cancellation_policy=abandonwithinstalled_runtime_activity_policy; native has no post-run authority for an abandoned activity.57e87238df33369a759099c1fb298a1246749e41,Workflow\V2\Support\ActivitySnapshot::fromExecution, emitsschedule_to_close_deadline_atonly when the activity cancellation policy is exactlyabandon.0005test's abandoned activity was outside the admitted native policy, and its assertion that the snapshot must contain that deadline was not a valid maintained-baseline requirement.0006correctly removes the unsupported policy and expects the field to be absent.Keep the independent production refusal for
abandon, and retain meaningful row/snapshot equality for fields that are part of admitted policies. Do not add native abandon admission merely to preserve the external reference test. This conclusion does not accept the cumulative0005source, its benchmark claims, or its known stale wait-ranking shortcut; the full0006wait-order correction and backend controls remain required.No supplied patch was applied, executed, compiled or benchmarked.
rmcdaniel commentedon Nov 8, 2026
Operator nudge: this benchmax ticket (P0) has had no linked PR and no activity for about 27 h. Please pick it up next, after the current work on #1057. Suggested order for the stalled server tickets: #1063, #1068, #1075, #1077, #1078, #1079, then #1081 (P2); this one is 4 of 7. Open a draft PR as soon as work starts so the tracker shows it. If it is blocked, add the status:blocked label with a one-line reason.
rmcdaniel commentedon Nov 9, 2026
General started the bounded #1077 source slice in draft #1115. The old successful-completion read is retained only as the test reference. Production uses the already-fenced activity row plus the exact written status/exception/close time; ordinary snapshot callers keep their read/decode order. Source 4613b8a/tree a9c6abef on actual main 5bde7ed; delta 14,980 B / 1d15e651838d783ed2c7c2da1686a5f61465364bb9d71728f2abcf56cb5503e0. All four database families use the same new named control via the existing source suites. First compile/test is hosted; no run, suite or benchmark PASS is claimed yet. Other cuts, wait-order controls and original WIP refs stay intact.
Count reference statement construction inside its measurement scope