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
//! What:
//! Where the landing's own syncs wait: not on the path the person waits on, unless the landing publishes nothing.
//! Why:
//! The owner's rule for the commit protocol, 2026-11-03: "Force at leaving, else background": no forced write while the person
//! waits, so a power cut loses that attempt exactly as it does with native Git (staged changes and files are intact, the
//! commit is run again), and whatever the command already promised when it left (an error, a withdrawal, a hand-over) was
//! forced to disk first. A landing that publishes a commit therefore only queues the syncs of its records (the intent, the
//! lock marker, the completion record, the done mark and the directory syncs); one background thread makes them in batches,
//! each batch overlapped. A landing that publishes nothing syncs what it wrote before it returns. Git's own object, reference
//! and index writes follow `core.fsync` as they do for native Git: the in-process index writer flushes only when
//! `core.fsync` names the index (`ConfigSnapshot::hardens_index`).
//! Nothing here changes what a killed process leaves: the page cache keeps every write whether or not it was synced.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const landing = beginDeferral({ hardensIndex });
//! // ... flushFile(file, 'intent-file') queues; flushDir(dir, 'intent-dir') queues ...
//! landing.settle(published); // true: hand the queue to the background thread; false: sync it now
//! ```
use std::cell::{Cell, RefCell};
use std::fs::File;
use std::sync::OnceLock;
use std::sync::mpsc::{Receiver, Sender, channel};
/// One sync waiting for its turn: a handle on the file or directory, and what it is for.
struct Pending {
/// A handle on the same file or directory the record lives in; a flush through it flushes the file itself.
handle: File,
/// What the sync is for, as the controls and the log name it.
what: &'static str,
}
/// What the background thread is asked to do.
enum Job {
/// Sync every handle of a landing, all at once.
Batch(Vec<Pending>),
/// Answer on the channel when every earlier job is done.
Barrier(Sender<()>),
}
thread_local! {
/// Whether the landing on this thread queues its syncs (`true`) or makes them where it asks (`false`).
static DEFERRING: Cell<bool> = const { Cell::new(false) };
/// Whether the index the landing writes is flushed before it is installed: `core.fsync` names the index, or nothing said.
static INDEX_HARDENED: Cell<bool> = const { Cell::new(true) };
/// The syncs this landing has queued so far.
static QUEUED: RefCell<Vec<Pending>> = const { RefCell::new(Vec::new()) };
}
/// The sender of the one background thread, started by the first landing that publishes.
static BACKGROUND: OnceLock<Sender<Job>> = OnceLock::new();
/// How many syncs the background thread has made, for a control that waits for them.
pub static BACKGROUND_SYNCED: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
/// Sync every handle of a batch at once, one thread each, and log what failed: nobody waits for this, so a failure cannot
/// fail a commit that was already answered, and the next landing or restart handles the record as it would after a power cut.
fn sync_batch(batch: Vec<Pending>) {
std::thread::scope(|scope| {
for pending in &batch {
scope.spawn(move || {
if let Err(error) = pending.handle.sync_all() {
crate::service_warn!("the background sync of the {} failed: {error}", pending.what);
}
BACKGROUND_SYNCED.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
});
}
});
}
/// The background thread: takes the jobs in order until the sender is gone.
fn background(jobs: &Receiver<Job>) {
while let Ok(job) = jobs.recv() {
match job {
Job::Batch(batch) => sync_batch(batch),
Job::Barrier(done) => {
let _answered = done.send(());
}
}
}
}
/// The sender of the background thread, started when first asked for.
fn sender() -> &'static Sender<Job> {
return BACKGROUND.get_or_init(|| {
let (sender, jobs) = channel::<Job>();
let started = std::thread::Builder::new().name(String::from("cli-git-durable")).spawn(move || { background(&jobs); });
if let Err(error) = started {
crate::service_warn!("the thread that makes the landing's syncs could not start ({error}); they are made where they are asked for");
}
return sender;
});
}
/// Waits until every sync handed to the background thread so far is done.
/// A control uses it to see the queue empty; the service does not wait for it on a commit's path.
pub fn drain() {
let (done, answered) = channel::<()>();
if sender().send(Job::Barrier(done)).is_ok() {
let _waited = answered.recv();
}
}
/// What a landing holds while it runs: its syncs are queued, and it settles them when it ends.
pub struct Deferral {
/// Whether `settle` has run; a deferral dropped without it (a panic) syncs what it queued.
settled: bool,
}
/// Starts queueing the syncs of the landing on this thread.
/// `hardens_index` says whether the index file the landing installs is flushed before the rename (`core.fsync` names it).
pub fn begin(hardens_index: bool) -> Deferral {
DEFERRING.with(|flag| { flag.set(true); });
INDEX_HARDENED.with(|flag| { flag.set(hardens_index); });
QUEUED.with(|queue| { queue.borrow_mut().clear(); });
return Deferral { settled: false };
}
/// Ends the deferral on this thread and returns what it queued.
fn end() -> Vec<Pending> {
DEFERRING.with(|flag| { flag.set(false); });
INDEX_HARDENED.with(|flag| { flag.set(true); });
return QUEUED.with(|queue| { return std::mem::take(&mut *queue.borrow_mut()); });
}
/// Syncs every handle now, all at once, and names each as waited for.
fn force(batch: Vec<Pending>) {
for pending in &batch {
crate::service_fault::note(format!("sync-waited:{}", pending.what).as_str());
}
std::thread::scope(|scope| {
for pending in &batch {
scope.spawn(move || {
if let Err(error) = pending.handle.sync_all() {
crate::service_warn!("the sync of the {} failed: {error}", pending.what);
}
});
}
});
}
impl Deferral {
/// Ends the deferral. A landing that published a commit hands its queue to the background thread and returns at once; a
/// landing that published nothing (a withdrawal, an error, a commit sent back) syncs its records before it returns.
pub fn settle(mut self, published: bool) {
self.settled = true;
let batch: Vec<Pending> = end();
if batch.is_empty() {
return;
}
if published {
if let Err(rejected) = sender().send(Job::Batch(batch)) {
// The thread is gone, so the queue is made here rather than lost.
if let Job::Batch(returned) = rejected.0 {
force(returned);
}
}
} else {
force(batch);
}
}
}
impl Drop for Deferral {
fn drop(&mut self) {
if !self.settled {
force(end());
}
}
}
/// Flushes a file that is a landing's record: queued while a landing defers, made now otherwise.
///
/// # Errors
///
/// The error of the flush when it is made now.
pub fn flush_file(file: &File, what: &'static str) -> std::io::Result<()> {
if DEFERRING.with(Cell::get) {
crate::service_fault::note(format!("sync-queued:{what}").as_str());
QUEUED.with(|queue| -> std::io::Result<()> {
queue.borrow_mut().push(Pending { handle: file.try_clone()?, what });
return Ok(());
})?;
return Ok(());
}
let synced: std::io::Result<()> = file.sync_all();
crate::service_fault::note(format!("sync-waited:{what}").as_str());
return synced;
}
/// Flushes a directory whose entries are a landing's records, as `flush_file` does for a file.
///
/// # Errors
///
/// The error of opening the directory, or of the flush when it is made now.
pub fn flush_directory(directory: &std::path::Path, what: &'static str) -> std::io::Result<()> {
return flush_file(&File::open(directory)?, what);
}
/// Flushes the index file the landing writes, when the index is hardened: `core.fsync` names it, as native Git's index write
/// is hardened then. Otherwise the index is written as native Git writes it, unflushed. It is made now, on the landing's path,
/// as Git's own write is.
///
/// # Errors
///
/// The error of the flush.
pub fn flush_index(file: &File, what: &'static str) -> std::io::Result<()> {
if INDEX_HARDENED.with(Cell::get) {
let synced: std::io::Result<()> = file.sync_all();
crate::service_fault::note(format!("sync-waited:{what}").as_str());
return synced;
}
return Ok(());
}
/// Whether the index file the landing writes is flushed before it is installed.
pub fn index_is_hardened() -> bool {
return INDEX_HARDENED.with(Cell::get);
}
/// Controls for the queue stay out of the release executable.
#[cfg(test)]
#[path = "service_durable_tests.rs"]
mod tests;