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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
//! What:
//! The owner record of one commit-transaction directory (`owner.json`,
//! schema version 2)
//! and the liveness rule that decides whether recovery may touch the transaction.
//! Why:
//! The owner record is written before the directory is published,
//! so recovery can skip a
//! live owner's transaction at every phase.
//! Both wrappers write and judge the same record
//! (`src/policy-engine/commit-transaction-owner.ts`).
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const owner = parseTransactionOwner(bytes); const liveness = await classifyTransactionOwner(owner);
//! ```
/// Record field readers,
/// the fatal decoder and the ordered writer.
use super::json_record::{
ObjectWriter, decode_fatal, number_equals, parse_object, safe_integer_field, string_field,
};
/// Birth identities and the existence probe.
use super::process_identity::{
ProcessIdentityError, current_birth_identity, process_birth_identity,
};
/// Owner record schema;
/// version 2 marks the per-transaction directory layout.
pub const TRANSACTION_OWNER_SCHEMA_VERSION: i64 = 2;
/// What:
/// The durable owner of one transaction directory.
/// Why:
/// The PID and birth identity prove liveness;
/// `created_at`,
/// the invocation start time
/// as ISO-8601 text,
/// orders dead transactions for recovery and grants the reservation.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type TransactionOwnerRecord = { schemaVersion: 2; transactionId: string; ownerPid: number; ownerIdentity: string; createdAt: string };
/// ```
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct TransactionOwner {
/// Transaction ID,
/// equal to the directory name.
pub transaction_id: String,
/// Wrapper process owning the transaction.
pub owner_pid: i64,
/// Process-birth identity of the owner.
pub owner_identity: String,
/// ISO-8601 invocation start time.
pub created_at: String,
}
/// What:
/// Whether a recorded owner still runs.
/// Why:
/// Only a dead owner's transaction may be recovered.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type TransactionOwnerLiveness = 'alive' | 'dead';
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OwnerLiveness {
/// The PID names the original process.
Alive,
/// The process exited or its PID was reused.
Dead,
}
/// What:
/// Why an owner's liveness could not be decided.
/// Why:
/// A refused probe (for example another account's process) must stop recovery rather
/// than let it touch a transaction that may be live,
/// as the incumbent's throw does.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// class CommitTransactionRecoveryError extends Error {}
/// ```
#[derive(Debug)]
pub enum OwnerLivenessError {
/// The existence probe was refused.
Probe(std::io::Error),
/// The birth identity could not be read.
Identity(ProcessIdentityError),
}
/// What:
/// The current process's owner record for a fresh transaction.
/// Why:
/// Written into the staging directory before publication.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// async function createTransactionOwnerRecord({ transactionId, createdAt }): Promise<TransactionOwnerRecord>;
/// ```
pub fn current_transaction_owner(
transaction_id: &str,
created_at: &str,
) -> Result<TransactionOwner, ProcessIdentityError> {
return Ok(TransactionOwner {
transaction_id: String::from(transaction_id),
owner_pid: i64::from(std::process::id()),
owner_identity: current_birth_identity()?,
created_at: String::from(created_at),
});
}
/// What:
/// Encode an owner record as compact JSON in the incumbent's field order,
/// with a newline.
/// Why:
/// `encodeTransactionOwner` writes exactly these bytes.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// `${JSON.stringify({ schemaVersion, transactionId, ownerPid, ownerIdentity, createdAt })}\n`
/// ```
pub fn encode_transaction_owner(owner: &TransactionOwner) -> String {
// `mut` allows appending one field at a time.
let mut writer: ObjectWriter = ObjectWriter::new();
writer
.integer("schemaVersion", TRANSACTION_OWNER_SCHEMA_VERSION)
.string("transactionId", owner.transaction_id.as_str())
.integer("ownerPid", owner.owner_pid)
.string("ownerIdentity", owner.owner_identity.as_str())
.string("createdAt", owner.created_at.as_str());
return format!("{}\n", writer.finish());
}
/// What:
/// Parse owner record bytes,
/// or nothing when they are not a valid record.
/// Why:
/// The incumbent's `parseTransactionOwner` decodes with a fatal UTF-8 decoder and accepts
/// schema version 2,
/// text `transactionId` (possibly empty,
/// which the directory-name check
/// then rejects),
/// a safe-integer PID of at least 1,
/// a non-empty identity and text
/// `createdAt`.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// function parseTransactionOwner(bytes: Uint8Array): TransactionOwnerRecord; // throws when malformed
/// ```
pub fn parse_transaction_owner(bytes: &[u8]) -> Option<TransactionOwner> {
let map: serde_json::Map<String, serde_json::Value> = parse_object(decode_fatal(bytes)?)?;
if !number_equals(&map, "schemaVersion", TRANSACTION_OWNER_SCHEMA_VERSION) {
return None;
}
let transaction_id: &str = string_field(&map, "transactionId")?;
let owner_pid: i64 = safe_integer_field(&map, "ownerPid")?;
let owner_identity: &str = string_field(&map, "ownerIdentity")?;
let created_at: &str = string_field(&map, "createdAt")?;
if owner_pid < 1 || owner_identity.is_empty() {
return None;
}
return Some(TransactionOwner {
transaction_id: String::from(transaction_id),
owner_pid,
owner_identity: String::from(owner_identity),
created_at: String::from(created_at),
});
}
/// What:
/// Whether the process `pid` exists,
/// as the incumbent's `processIsAlive` asks first.
/// On systems without signals,
/// existence is left to the identity probe.
/// Why:
/// `kill(pid, 0)` answering `ESRCH` is a dead owner;
/// any other refusal is reported.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// function processIsAlive(pid: number): boolean;
/// ```
fn process_exists(pid: i64) -> Result<bool, OwnerLivenessError> {
#[cfg(unix)]
{
return super::process_identity::signal_probe(pid).map_err(OwnerLivenessError::Probe);
}
#[cfg(not(unix))]
{
let _ = pid;
return Ok(true);
}
}
/// What:
/// Classify a recorded owner:
/// alive only while its PID still names the same process birth.
/// Why:
/// An exited process,
/// a zombie,
/// or a younger process reusing the PID is a dead owner.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// async function classifyTransactionOwner({ ownerPid, ownerIdentity }): Promise<'alive' | 'dead'>;
/// ```
pub fn classify_transaction_owner(
owner_pid: i64,
owner_identity: &str,
) -> Result<OwnerLiveness, OwnerLivenessError> {
if !process_exists(owner_pid)? {
return Ok(OwnerLiveness::Dead);
}
match process_birth_identity(owner_pid).map_err(OwnerLivenessError::Identity)? {
Some(current) if current == owner_identity => return Ok(OwnerLiveness::Alive),
Some(_) | None => return Ok(OwnerLiveness::Dead),
}
}
/// Owner record and liveness controls stay out of the release executable.
#[cfg(test)]
#[path = "transaction_owner_tests.rs"]
mod tests;