1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
// SPDX-License-Identifier: BUSL-1.1
use crate::bridge::envelope::{PhysicalPlan, Request};
use crate::types::{Lsn, RequestId};
/// An execution task on a Data Plane core.
///
/// Wraps a `Request` envelope with execution state tracking.
/// This type is `!Send` — it lives and dies on a single core.
pub struct ExecutionTask {
/// The original request envelope.
pub request: Request,
/// Current execution state.
pub state: TaskState,
/// WAL LSN the Control Plane allocated for this write at wal-dispatch time,
/// carried alongside the request so the apply chokepoint can record the
/// per-key / per-collection write version (see
/// [`CoreLoop::note_write_lsn`](crate::data::executor::core_loop::CoreLoop)).
/// `None` for reads, control ops, and writes whose LSN is not (yet) threaded
/// — the version index is skipped rather than advanced with a wrong value.
/// Copied from [`Request::wal_lsn`](crate::bridge::envelope::Request) in
/// [`ExecutionTask::new`] — the request envelope is the cross-plane channel
/// that carries the allocated LSN from Control Plane to this core.
pub wal_lsn: Option<Lsn>,
/// Wall-clock instant (ms since epoch) the Control Plane resolved at
/// WAL-append time for a TTL-bearing KV write, carried alongside the
/// request so live apply installs the SAME instant the durable WAL record
/// carries rather than re-reading the clock. Re-reading it at apply time
/// would let the live value disagree with the durable one by the dispatch
/// latency — harmless day to day, but a crash between the two would have
/// replay recompute `now_ms` at restart time instead of installing the
/// original instant, pushing the TTL's expiry forward by the
/// crash-to-restart delay. `None` for non-TTL writes, reads, and writes
/// whose resolved instant is not (yet) threaded.
/// Copied from [`Request::resolved_now_ms`](crate::bridge::envelope::Request)
/// in [`ExecutionTask::new`], same as `wal_lsn`.
pub resolved_now_ms: Option<u64>,
}
/// Lifecycle states for a Data Plane task.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TaskState {
/// Queued, waiting for core capacity.
Pending,
/// Actively executing (io_uring submitted, SIMD running, etc.)
Running,
/// Completed successfully — response ready to send back.
Completed,
/// Cancelled by Control Plane.
Cancelled,
/// Failed with error.
Failed,
}
impl ExecutionTask {
pub fn new(request: Request) -> Self {
// The request envelope is the only Control->Data channel: copy the
// allocated write LSN and resolved TTL instant off it so the apply
// chokepoints on this core can record the per-key / per-collection
// write version and install the same expiry instant the WAL record
// carries.
let wal_lsn = request.wal_lsn;
let resolved_now_ms = request.resolved_now_ms;
Self {
request,
state: TaskState::Pending,
wal_lsn,
resolved_now_ms,
}
}
/// Construct a task carrying the WAL LSN allocated for its write.
/// `resolved_now_ms` is `None` — callers needing a resolved TTL instant on
/// a directly-constructed task should go through [`ExecutionTask::new`].
pub fn with_wal_lsn(request: Request, wal_lsn: Option<Lsn>) -> Self {
Self {
request,
state: TaskState::Pending,
wal_lsn,
resolved_now_ms: None,
}
}
/// WAL LSN allocated for this write, if any.
pub fn wal_lsn(&self) -> Option<Lsn> {
self.wal_lsn
}
/// Wall-clock instant the Control Plane resolved for a TTL-bearing KV
/// write, if any. See the field doc on [`ExecutionTask::resolved_now_ms`].
pub fn resolved_now_ms(&self) -> Option<u64> {
self.resolved_now_ms
}
pub fn request_id(&self) -> RequestId {
self.request.request_id
}
pub fn plan(&self) -> &PhysicalPlan {
&self.request.plan
}
pub fn is_expired(&self) -> bool {
std::time::Instant::now() > self.request.deadline
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::bridge::envelope::{PhysicalPlan, Priority};
use crate::types::{DatabaseId, ReadConsistency, TenantId, TraceId, VShardId};
use nodedb_physical::physical_plan::MetaOp;
use std::time::{Duration, Instant};
fn request_with_wal_lsn(wal_lsn: Option<Lsn>) -> Request {
Request {
request_id: RequestId::new(1),
tenant_id: TenantId::new(1),
database_id: DatabaseId::DEFAULT,
vshard_id: VShardId::new(0),
plan: PhysicalPlan::Meta(MetaOp::Cancel {
target_request_id: RequestId::new(7),
}),
deadline: Instant::now() + Duration::from_secs(5),
priority: Priority::Normal,
trace_id: TraceId::ZERO,
consistency: ReadConsistency::Strong,
idempotency_key: None,
event_source: crate::event::EventSource::User,
user_roles: Vec::new(),
user_id: None,
statement_digest: None,
txn_id: None,
wal_lsn,
resolved_now_ms: None,
admission: crate::bridge::envelope::Admission::Exempt(
crate::bridge::envelope::ExemptReason::Read,
),
}
}
#[test]
fn new_copies_request_wal_lsn_onto_task() {
// The request envelope is the only Control->Data channel for the
// allocated write LSN: `ExecutionTask::new` must copy it so the apply
// chokepoints can record the committed write version.
let lsn = Lsn::new(4242);
let task = ExecutionTask::new(request_with_wal_lsn(Some(lsn)));
assert_eq!(task.wal_lsn(), Some(lsn));
}
#[test]
fn new_leaves_wal_lsn_none_for_reads() {
let task = ExecutionTask::new(request_with_wal_lsn(None));
assert_eq!(task.wal_lsn(), None);
}
#[test]
fn new_copies_request_resolved_now_ms_onto_task() {
let mut request = request_with_wal_lsn(Some(Lsn::new(1)));
request.resolved_now_ms = Some(1_000);
let task = ExecutionTask::new(request);
assert_eq!(task.resolved_now_ms(), Some(1_000));
}
#[test]
fn with_wal_lsn_leaves_resolved_now_ms_none() {
let task = ExecutionTask::with_wal_lsn(request_with_wal_lsn(None), Some(Lsn::new(5)));
assert_eq!(task.resolved_now_ms(), None);
}
}