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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
//! What:
//! The lines the resident commit service writes to its log (standard error,
//! which the service manager keeps in the journal)
//! about its own life and about the time each request took:
//! when it started,
//! one line for each request it answered with the
//! phases of the wait,
//! a retire request,
//! and its exit with the reason and the uptime.
//! Why:
//! After a slow commit the journal held `ready` and nothing else,
//! so nothing said whether the time went to the queue,
//! to
//! another program's hold on the index,
//! to the landing or to a service being replaced.
//! These lines cost one clock read per
//! phase edge and one line per request;
//! they name no person (the commit's message and identities never appear) and are not
//! shown to the person at the terminal.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! console.error(`cli-git-service: answered request=${id} queued_ms=${queued} lock_wait_ms=${lockWait} total_ms=${total}`);
//! ```
use super::service_index_found::WaitNotes;
use super::journald_native::{Delivery, MESSAGE_LIMIT, Priority, Sink, StreamWriter, cut, one_line};
use super::lfs_output_mask::mask_userinfo_text;
use std::sync::OnceLock;
use std::time::{Duration, Instant};
/// The prefix of every line the service writes,
/// which `journalctl` shows after the unit name.
pub const LOG_PREFIX: &str = "cli-git-service: ";
/// When a connection was accepted and how long its request line took to arrive.
#[derive(Clone, Copy, Debug)]
pub struct Arrival {
/// The moment the connection was accepted.
pub accepted: Instant,
/// Time from the acceptance to the whole request line being read;
/// a client that is slow to send shows up here and not as a
/// wait in the queue.
pub read: Duration,
}
/// What:
/// The phases of one landing the service itself spent time in,
/// summed over the attempts a moving branch made it repeat.
/// Why:
/// The wait for another program's hold on the index,
/// the objects,
/// the journal write,
/// the branch move and the index update are
/// the stretches a slow landing can lose time in;
/// whatever else a landing did is reported as the reads that remain.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type LandTiming = { lockWait: number; lockHolder?: string; objects: number; journal: number; refUpdate: number; sync: number; attempts: number };
/// ```
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct LandTiming {
/// Time to take the index lock,
/// which is the whole wait for a foreign holder.
pub lock_wait: Duration,
/// The holder the evidence named while the lock was waited for,
/// when it named one.
pub lock_holder: Option<String>,
/// What the acquisition noted about its wait:
/// what the first and the last look at the held lock found,
/// how many looks there were,
/// and how the time split between flushing the marker and waiting for the holder.
pub lock_notes: WaitNotes,
/// Time to make the commits:
/// one import for the batch,
/// or the checks of a commit Git built.
pub objects: Duration,
/// Time to write the intent to the journal.
pub journal: Duration,
/// Time of the compare-and-swap that moves the branch.
pub ref_update: Duration,
/// Time to remove the conclusion state and to build and install the new index.
pub sync: Duration,
/// How long the index lock was held, from its placement to its release: the branch move and the index build.
/// Not a phase of its own (it overlaps `ref_update` and `sync`), so it is not part of what `reads` subtracts.
pub hold: Duration,
/// Passes over the batch:
/// one,
/// and one more each time the branch moved under it.
pub attempts: usize,
}
impl LandTiming {
/// What `landing` spent outside the named phases:
/// the branch reads,
/// the comparisons with the branch,
/// the check for removed
/// content.
pub fn reads(&self, landing: Duration) -> Duration {
return landing.saturating_sub(self.lock_wait + self.objects + self.journal + self.ref_update + self.sync);
}
}
/// Everything the line for one answered request states.
pub struct Answered<'record> {
/// The request's identity.
pub request: &'record str,
/// The reference the request names.
pub reference: &'record str,
/// What became of it:
/// a word such as `landed`,
/// `already`,
/// `retry-prepare`,
/// `branch-moved` or `failed:<code>`.
pub outcome: &'record str,
/// The batch number,
/// shared by every request landed together.
pub batch: u64,
/// How many requests the batch of this branch held.
pub size: usize,
/// This request's place among them,
/// from one.
pub position: usize,
/// How the connection and its request line arrived.
pub arrival: Arrival,
/// Time from the request being read to its batch starting to land.
pub queued: Duration,
/// Time to settle the batches a killed service left open,
/// which runs before each batch.
pub settle: Duration,
/// The phases of the landing.
pub landing: &'record LandTiming,
/// Time the landing took,
/// from its start to its return.
pub land_total: Duration,
/// Time from the landing's end to this request being answered.
pub answer: Duration,
/// Time from the acceptance of the connection to the answer.
pub total: Duration,
}
/// The word for a named holder,
/// in the quoted form.
fn holder_field(who: &String) -> String {
return format!(" lock_holder={who:?}");
}
/// The words for what the wait for the lock found,
/// empty when the lock was free at the first placement:
/// the placements that found it held,
/// the time spent flushing the marker and the time spent waiting,
/// and the first and the last description of the held lock.
pub fn lock_notes_fields(notes: &WaitNotes) -> String {
if notes.looks == 0 {
return format!(" lock_stage_ms={} lock_stage_kind={} lock_place_ms={}", notes.staging.as_millis(), notes.stage_kind, notes.placing.as_millis());
}
let first: &str = notes.found.as_deref().unwrap_or("none");
let last: &str = notes.last.as_deref().unwrap_or("none");
return format!(
" lock_stage_ms={} lock_stage_kind={} lock_place_ms={} lock_looks={} lock_found={first:?} lock_last={last:?}",
notes.staging.as_millis(),
notes.stage_kind,
notes.placing.as_millis(),
notes.looks
);
}
/// The wait for the index lock that is written down even when no holder was met:
/// a flush of the marker or a wait that long is a slow landing's first suspect.
pub const SLOW_LOCK_STEP: Duration = Duration::from_secs(1);
/// What:
/// The line for the end of a wait for the index lock,
/// or nothing when the lock was free at the first placement and the acquisition was quick.
/// Why:
/// Written when the wait ends,
/// not with the answer,
/// so a landing that fails afterwards still leaves what it waited for.
/// It says the total,
/// how it split between flushing the marker and waiting for a holder,
/// how many times the lock was found held,
/// and what the first and the last look found (`lock_found`,
/// `lock_last`).
///
/// In TS you'd write (pseudocode):
/// ```ts
/// waitLine(seq, total, notes, taken) // 'waited batch=7 result=taken lock_wait_ms=5629 ...' | undefined
/// ```
pub fn wait_line(seq: u64, total: Duration, notes: &WaitNotes, taken: bool) -> Option<String> {
if notes.looks == 0 && notes.staging < SLOW_LOCK_STEP && total < SLOW_LOCK_STEP {
return None;
}
return Some(format!(
"waited for the index lock batch={seq} result={} lock_wait_ms={}{}",
if taken { "taken" } else { "refused" },
total.as_millis(),
lock_notes_fields(notes)
));
}
/// What:
/// One answered request as a line of `key=value` words,
/// without the prefix.
/// Why:
/// Text values are written in Rust's quoted form,
/// which has no raw newline or unescaped quote,
/// so the line stays one line
/// whatever a client sent;
/// the numbers are whole milliseconds.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// `answered request=${JSON.stringify(id)} outcome=${outcome} queued_ms=${queued} ...`
/// ```
pub fn answered_line(answered: &Answered<'_>) -> String {
let mut holder: String = answered.landing.lock_holder.as_ref().map_or_else(String::new, holder_field);
holder.push_str(lock_notes_fields(&answered.landing.lock_notes).as_str());
return format!(
"answered request={:?} branch={:?} outcome={} batch={} size={} position={} read_ms={} queued_ms={} settle_ms={} \
lock_wait_ms={}{holder} lock_hold_ms={} reads_ms={} objects_ms={} journal_ms={} ref_ms={} sync_ms={} attempts={} landing_ms={} \
answer_ms={} total_ms={}",
answered.request,
answered.reference,
answered.outcome,
answered.batch,
answered.size,
answered.position,
answered.arrival.read.as_millis(),
answered.queued.as_millis(),
answered.settle.as_millis(),
answered.landing.lock_wait.as_millis(),
answered.landing.hold.as_millis(),
answered.landing.reads(answered.land_total).as_millis(),
answered.landing.objects.as_millis(),
answered.landing.journal.as_millis(),
answered.landing.ref_update.as_millis(),
answered.landing.sync.as_millis(),
answered.landing.attempts,
answered.land_total.as_millis(),
answered.answer.as_millis(),
answered.total.as_millis(),
);
}
/// The line a service writes as it starts,
/// before `ready`:
/// the release it announces,
/// the protocol,
/// its process and the worktree
/// instance it serves.
pub fn started_line(release: &str, protocol: u32, instance: Option<&str>) -> String {
return format!("started release={release:?} protocol={protocol} pid={} instance={}", std::process::id(), instance.unwrap_or("any"));
}
/// The line for a retire message:
/// a client of another release found this service and asked it to finish its batch and exit.
pub fn retire_line(release: &str) -> String {
return format!("retire requested by a client; this service is release {release:?} and exits after its batch");
}
/// Why a service ended.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ExitReason {
/// Nothing to do for the whole idle period.
Idle,
/// A client asked it to retire.
Retire,
/// The queue its connections feed closed,
/// which only happens when the acceptor thread died.
Error,
}
impl ExitReason {
/// The word written in the line.
pub fn word(self) -> &'static str {
return match self {
ExitReason::Idle => "idle",
ExitReason::Retire => "retire",
ExitReason::Error => "error",
};
}
}
/// The line for the service's own exit,
/// with the time it ran since it began serving.
pub fn exit_line(reason: ExitReason, uptime: Duration) -> String {
return format!("exiting reason={} uptime_ms={}", reason.word(), uptime.as_millis());
}
/// The wall clock in whole milliseconds since the Unix epoch,
/// the time a log line states for its own event.
pub fn wall_millis() -> u128 {
return std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |elapsed| { return elapsed.as_millis(); });
}
/// What:
/// The text of one log line:
/// the prefix,
/// the line,
/// the time of the event as `at_ms=<epoch milliseconds>` at the end,
/// and the newline.
/// Why:
/// The journal stamps a line when it receives it,
/// and under load that can be seconds after the event;
/// the line states its own time,
/// at the end so that the prefix and the first word stay where readers look for them.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// `${LOG_PREFIX}${line} at_ms=${Date.now()}\n`
/// ```
pub fn stamped_line(line: &str, at_ms: u128) -> String {
return format!("{LOG_PREFIX}{line} at_ms={at_ms}\n");
}
/// What:
/// Writes one line to the service's log.
/// Why:
/// Standard error is the service manager's journal;
/// one `write` carries the prefix,
/// the line,
/// its own time and the newline together,
/// and a
/// closed or broken stream is not the service's failure.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// console.error(LOG_PREFIX + line + ` at_ms=${Date.now()}`)
/// ```
pub fn log(line: &str) {
log_entry(Priority::Info, line, &[]);
}
/// What the service says of itself in every entry: its release and, when it serves one worktree, its instance.
static CONTEXT: OnceLock<Vec<(&'static str, String)>> = OnceLock::new();
/// Where entries go, decided at the first one.
static SINK: OnceLock<Sink> = OnceLock::new();
/// The sink of this process.
pub fn sink() -> &'static Sink {
return SINK.get_or_init(Sink::from_environment);
}
/// Names the release and instance every later entry carries as `CLI_GIT_RELEASE` and `CLI_GIT_INSTANCE`; the first call wins.
pub fn set_context(release: &str, instance: Option<&str>) {
let mut fields: Vec<(&'static str, String)> = vec![("CLI_GIT_RELEASE", release.to_string())];
if let Some(name) = instance {
fields.push(("CLI_GIT_INSTANCE", name.to_string()));
}
let _first = CONTEXT.set(fields);
}
/// What:
/// Writes one entry of a priority, with fields a reader can filter the journal on, and gives where it went.
/// Why:
/// The userinfo of every URL is masked (a push or a review reports what Git and a server said), the line on standard error
/// is one line,
/// then it states its own time (`stamped_line`); a native entry also carries the time as `CLI_GIT_AT_MS`, the priority
/// and the context. Nothing here waits: see `journald_native`.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// logEntry('warning', 'cannot listen', [['CLI_GIT_REQUEST_ID', id]])
/// ```
pub fn log_entry(priority: Priority, line: &str, fields: &[(&str, String)]) -> Delivery {
let at_ms: u128 = wall_millis();
// Masked before it is cut: a cut that falls just before the `@` of `scheme://user:secret@host` would leave a password
// that no longer looks like userinfo.
let masked: String = mask_userinfo_text(line);
let mut all: Vec<(&str, String)> = CONTEXT.get().map_or_else(Vec::new, |context| { return context.iter().map(|(name, value)| { return (*name, value.clone()); }).collect(); });
all.push(("CLI_GIT_AT_MS", at_ms.to_string()));
all.extend(fields.iter().map(|(name, value)| { return (*name, mask_userinfo_text(value.as_str())); }));
// A native entry carries the text as written (a newline in the length-prefixed form); the line on standard error is one line.
let message: String = format!("{} at_ms={at_ms}", cut(masked.as_str(), MESSAGE_LIMIT));
#[cfg(test)]
if capture::record(priority, message.as_str(), all.as_slice()) {
return Delivery::Stderr;
}
let text: String = stamped_line(one_line(masked.as_str(), MESSAGE_LIMIT).as_str(), at_ms);
// On the journal's stream a line that finds no room is dropped, never waited for.
if sink().on_stream() {
return sink().deliver(priority, message.as_str(), all.as_slice(), text.as_str(), &mut StreamWriter::stderr());
}
return sink().deliver(priority, message.as_str(), all.as_slice(), text.as_str(), &mut std::io::stderr());
}
/// What:
/// Writes one formatted line at the `warning` priority, with its time.
/// Why:
/// Something did not go as intended and the service went on; `journalctl --priority=warning` shows it.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// serviceWarn(`the review of ${oid} was skipped: ${reason}`)
/// ```
#[macro_export]
macro_rules! service_warn {
($($argument:tt)*) => {
{
let _delivered = $crate::service_timing::log_entry($crate::journald_native::Priority::Warning, format!($($argument)*).as_str(), &[]);
}
};
}
/// What:
/// Writes one formatted line at the `err` priority, with its time.
/// Why:
/// Something the service needed failed; `journalctl --priority=err` shows it.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// serviceError(`cannot listen: ${error}`)
/// ```
#[macro_export]
macro_rules! service_error {
($($argument:tt)*) => {
{
let _delivered = $crate::service_timing::log_entry($crate::journald_native::Priority::Error, format!($($argument)*).as_str(), &[]);
}
};
}
/// What:
/// Writes one formatted line to the service's log,
/// with its time.
/// Why:
/// Every line the service writes states its own time (`stamped_line`);
/// the macro is the one way to write a line with arguments.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// serviceLog(`cannot take the index lock ${path}: ${error}`)
/// ```
#[macro_export]
macro_rules! service_log {
($($argument:tt)*) => {
$crate::service_timing::log(format!($($argument)*).as_str())
};
}
/// What a control sees of the entries the code under test writes, on the thread that runs it.
#[cfg(test)]
pub mod capture {
use super::Priority;
use std::cell::RefCell;
/// One entry as written: its priority, its message with the time, and its fields.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Captured {
/// How serious the entry is.
pub priority: Priority,
/// The message as the journal would show it.
pub message: String,
/// The fields beside the message.
pub fields: Vec<(String, String)>,
}
thread_local! {
static ENTRIES: RefCell<Option<Vec<Captured>>> = const { RefCell::new(None) };
}
/// Notes an entry when a control is capturing on this thread; says whether it did.
pub fn record(priority: Priority, message: &str, fields: &[(&str, String)]) -> bool {
return ENTRIES.with(|entries| {
return entries.borrow_mut().as_mut().is_some_and(|taken| {
taken.push(Captured { priority, message: message.to_string(), fields: fields.iter().map(|(name, value)| { return ((*name).to_string(), value.clone()); }).collect() });
return true;
});
});
}
/// Runs `body` and gives its entries, which are not written anywhere.
pub fn entries_of(body: impl FnOnce()) -> Vec<Captured> {
ENTRIES.with(|entries| { *entries.borrow_mut() = Some(Vec::new()); });
body();
return ENTRIES.with(|entries| { return entries.borrow_mut().take().unwrap_or_default(); });
}
}
/// The lines,
/// their fields,
/// and the escaping.
#[cfg(test)]
#[path = "service_timing_tests.rs"]
mod tests;