Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions crates/graphforge-api/src/repository.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2533,6 +2533,30 @@ mod tests {
}
}

/// Assert `writer.lock` and `checkpoints.lock` are free for exclusive acquire.
///
/// Does not touch generation `lease.lock` files; a live GraphForge may hold those.
fn assert_project_mutation_locks_free(state_path: &Path, phase: &str) {
let lock_root = state_path.join("locks");
for name in ["writer.lock", "checkpoints.lock"] {
let path = lock_root.join(name);
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)
.unwrap_or_else(|error| panic!("{phase}: open {name}: {error}"));
assert!(
FileExt::try_lock_exclusive(&file)
.unwrap_or_else(|error| panic!("{phase}: try_lock {name}: {error}")),
"{phase}: {name} was still held (issue #275 probe)"
);
FileExt::unlock(&file)
.unwrap_or_else(|error| panic!("{phase}: unlock {name}: {error}"));
}
}

fn test_skill_manifest() -> (Vec<u8>, [SkillBundleFile<'static>; 2]) {
let files = [
SkillBundleFile {
Expand Down Expand Up @@ -3575,6 +3599,13 @@ mod tests {
assert_eq!(preview.source_generation_uuid, first.generation_uuid);
assert_eq!(preview.current_generation_uuid, second.generation_uuid);

// Issue #275: if WriterBusy fires on an isolated project, distinguish a
// pre-existing held lock (after diff/preview) from an acquire-path bug.
assert_project_mutation_locks_free(
&context.state_path,
"after diff_checkpoints and preview_revert_to_checkpoint",
);

let generation_count = || {
fs::read_dir(context.state_path.join("generations"))
.unwrap()
Expand Down Expand Up @@ -3603,12 +3634,19 @@ mod tests {
.unwrap();
assert_eq!(snapshot.operation_uuid, first_operation);

assert_project_mutation_locks_free(&context.state_path, "after first revert_to_checkpoint");

let replay_receipt = graph.revert_to_checkpoint(request).unwrap();
assert_eq!(replay_receipt.schema, first_receipt.schema);
assert_eq!(replay_receipt.batches[0].num_rows(), 1);
assert_eq!(generation_count(), before_revert_count + 1);
drop(graph);

assert_project_mutation_locks_free(
&context.state_path,
"after drop of GraphForge that performed revert",
);

let reopened = crate::GraphForge::new(Some(context.state_path.to_str().unwrap())).unwrap();
let checkpoints = reopened
.list_checkpoints(crate::ListCheckpointsRequest::default())
Expand Down
244 changes: 244 additions & 0 deletions crates/graphforge-storage/src/project_checkpoints.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1821,6 +1821,153 @@ mod tests {
}
}

#[derive(Clone, Copy, Debug)]
enum CheckpointLockMode {
Shared,
Exclusive,
}

/// Hold `checkpoints.lock` (shared or exclusive) while `action` runs.
///
/// Writer is not held — the issue #275 CI signature is writer free +
/// checkpoints contended.
fn while_checkpoint_lock_is_held<T>(
root: &Path,
mode: CheckpointLockMode,
action: impl FnOnce() -> T,
) -> T {
let checkpoint_path = root.join(LOCKS_DIR).join(CHECKPOINT_LOCK_FILE);
let worker_path = checkpoint_path.clone();
let (ready_sender, ready_receiver) = mpsc::sync_channel(0);
let (release_sender, release_receiver) = mpsc::sync_channel(0);
let worker = std::thread::Builder::new()
.name("checkpoint-lock-holder".into())
.spawn(move || {
let checkpoint =
open_regular_lock(&worker_path).expect("phase=holder open checkpoints.lock");
let acquired = match mode {
CheckpointLockMode::Shared => FileExt::try_lock_shared(&checkpoint)
.expect("phase=holder acquire shared checkpoints.lock"),
CheckpointLockMode::Exclusive => FileExt::try_lock_exclusive(&checkpoint)
.expect("phase=holder acquire exclusive checkpoints.lock"),
};
assert!(
acquired,
"phase=holder checkpoints.lock unexpectedly busy mode={mode:?}"
);
ready_sender.send(()).expect("phase=holder publish ready");
release_receiver.recv().expect("phase=holder await release");
FileExt::unlock(&checkpoint).expect("phase=holder release checkpoints.lock");
})
.expect("phase=holder spawn");
let holder = WriterLockHolder {
release: Some(release_sender),
worker: Some(worker),
};
if let Err(error) = ready_receiver.recv_timeout(TEST_DEADLINE) {
drop(ready_receiver);
let cleanup = holder.finish();
panic!("phase=main await held checkpoints.lock error={error}; cleanup={cleanup:?}");
}
let result = catch_unwind(AssertUnwindSafe(action));
let cleanup = holder.finish();
match result {
Ok(value) => {
cleanup.unwrap_or_else(|error| panic!("phase=main holder cleanup error={error}"));
value
}
Err(original) => {
let _ = cleanup;
resume_unwind(original);
}
}
}

/// Reproduce the issue #275 schedule: acquire writer then checkpoints, unlock
/// writer early, leave checkpoints held while `action` runs.
fn while_checkpoint_held_after_writer_released<T>(
root: &Path,
action: impl FnOnce() -> T,
) -> T {
let lock_root = root.join(LOCKS_DIR);
let writer_path = lock_root.join(WRITER_LOCK_FILE);
let checkpoint_path = lock_root.join(CHECKPOINT_LOCK_FILE);
let worker_writer = writer_path.clone();
let worker_checkpoint = checkpoint_path.clone();
let (ready_sender, ready_receiver) = mpsc::sync_channel(0);
let (release_sender, release_receiver) = mpsc::sync_channel(0);
let worker = std::thread::Builder::new()
.name("checkpoint-early-writer-release".into())
.spawn(move || {
let writer =
open_regular_lock(&worker_writer).expect("phase=holder open writer.lock");
assert!(
FileExt::try_lock_exclusive(&writer).expect("phase=holder acquire writer.lock"),
"phase=holder writer.lock unexpectedly busy"
);
let checkpoint = open_regular_lock(&worker_checkpoint)
.expect("phase=holder open checkpoints.lock");
assert!(
FileExt::try_lock_exclusive(&checkpoint)
.expect("phase=holder acquire checkpoints.lock"),
"phase=holder checkpoints.lock unexpectedly busy"
);
FileExt::unlock(&writer).expect("phase=holder early release writer.lock");
ready_sender.send(()).expect("phase=holder publish ready");
release_receiver.recv().expect("phase=holder await release");
FileExt::unlock(&checkpoint).expect("phase=holder release checkpoints.lock");
})
.expect("phase=holder spawn");
let holder = WriterLockHolder {
release: Some(release_sender),
worker: Some(worker),
};
if let Err(error) = ready_receiver.recv_timeout(TEST_DEADLINE) {
drop(ready_receiver);
let cleanup = holder.finish();
panic!(
"phase=main await early-writer-release schedule error={error}; cleanup={cleanup:?}"
);
}
let result = catch_unwind(AssertUnwindSafe(action));
let cleanup = holder.finish();
match result {
Ok(value) => {
cleanup.unwrap_or_else(|error| panic!("phase=main holder cleanup error={error}"));
value
}
Err(original) => {
let _ = cleanup;
resume_unwind(original);
}
}
}

fn assert_mutation_locks_free(root: &Path, phase: &str) {
let lock_root = root.join(LOCKS_DIR);
for name in [WRITER_LOCK_FILE, CHECKPOINT_LOCK_FILE] {
let lock = open_regular_lock(&lock_root.join(name)).unwrap();
assert!(
FileExt::try_lock_exclusive(&lock).unwrap(),
"{phase}: {name} leaked"
);
FileExt::unlock(&lock).unwrap();
}
}

fn assert_checkpoint_mutation_busy_message(error: &GfError) {
assert_eq!(error.code(), "GF_WRITER_BUSY");
match error {
GfError::Project { message, .. } => {
assert_eq!(
message, "checkpoint mutation could not acquire checkpoints.lock",
"expected issue #275 WriterBusy message, got {message}"
);
}
other => panic!("expected Project WriterBusy, got {other}"),
}
}

struct BoundedChild {
child: std::process::Child,
reaped: bool,
Expand Down Expand Up @@ -2255,6 +2402,103 @@ mod tests {
drop(retained);
}

fn expect_mutation_locks_busy(root: &Path) -> GfError {
match acquire_mutation_locks(root) {
Ok(_locks) => panic!("expected WriterBusy acquiring mutation locks"),
Err(error) => error,
}
}

fn ensure_mutation_lock_files(root: &Path) {
let locks = acquire_mutation_locks(root).unwrap();
drop(locks);
}

#[test]
fn shared_checkpoint_reader_blocks_mutation_with_issue_275_message() {
let directory = tempdir().unwrap();
crate::open_or_initialize_project(directory.path()).unwrap();
ensure_mutation_lock_files(directory.path());
let error =
while_checkpoint_lock_is_held(directory.path(), CheckpointLockMode::Shared, || {
expect_mutation_locks_busy(directory.path())
});
assert_checkpoint_mutation_busy_message(&error);
}

#[test]
fn exclusive_checkpoint_holder_without_writer_blocks_mutation_with_issue_275_message() {
// Models revert post-handoff / recovery-style windows: writer free,
// checkpoints.lock still exclusive.
let directory = tempdir().unwrap();
crate::open_or_initialize_project(directory.path()).unwrap();
ensure_mutation_lock_files(directory.path());
let error =
while_checkpoint_lock_is_held(directory.path(), CheckpointLockMode::Exclusive, || {
expect_mutation_locks_busy(directory.path())
});
assert_checkpoint_mutation_busy_message(&error);
}

#[test]
fn early_writer_release_while_checkpoint_held_produces_issue_275_message() {
let directory = tempdir().unwrap();
crate::open_or_initialize_project(directory.path()).unwrap();
ensure_mutation_lock_files(directory.path());
let error = while_checkpoint_held_after_writer_released(directory.path(), || {
expect_mutation_locks_busy(directory.path())
});
assert_checkpoint_mutation_busy_message(&error);
}

#[test]
fn open_list_then_mutation_leaves_checkpoint_locks_free() {
// Rules out a same-thread shared-lock leak on the #275 failing sequence
// (open/list/read then immediate mutation).
let directory = tempdir().unwrap();
crate::open_or_initialize_project(directory.path()).unwrap();
create_checkpoint(
directory.path(),
&create_request(Uuid::from_u128(275), "Before"),
)
.unwrap();
let (_row, opened) = open_checkpoint_generation(directory.path(), "Before").unwrap();
drop(opened);
let listed = list_checkpoints(directory.path()).unwrap();
assert_eq!(listed.len(), 1);
assert_mutation_locks_free(
directory.path(),
"after open_checkpoint_generation and list",
);

let locks = acquire_mutation_locks(directory.path()).unwrap();
drop(locks);
assert_mutation_locks_free(directory.path(), "after uncontended acquire_mutation_locks");

create_checkpoint(
directory.path(),
&create_request(Uuid::from_u128(276), "After"),
)
.unwrap();
assert_mutation_locks_free(directory.path(), "after second create_checkpoint");
}

#[test]
fn recover_project_transactions_releases_checkpoint_before_returning() {
let directory = tempdir().unwrap();
crate::open_or_initialize_project(directory.path()).unwrap();
create_checkpoint(
directory.path(),
&create_request(Uuid::from_u128(277), "Retained"),
)
.unwrap();
crate::recover_project_transactions(directory.path()).unwrap();
assert_mutation_locks_free(
directory.path(),
"after recover_project_transactions return",
);
}

#[test]
fn create_list_delete_and_replays_are_deterministic() {
let directory = tempdir().unwrap();
Expand Down
4 changes: 4 additions & 0 deletions crates/graphforge-storage/src/project_recovery.rs
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,10 @@ pub fn recover_project_transactions(
recover_journals(&root, &transactions_root, &selected, &retained, &mut report)?;
}
report.preserved_unknown_entries += count_unknown_generation_entries(&root, &retained)?;
// Release checkpoints.lock before writer.lock so concurrent mutators never
// observe the issue #275 signature (writer free, checkpoints still held).
// Matches MutationLocks::Drop order in project_checkpoints.
drop(checkpoint_roots);
drop(writer_lock);
Ok(report)
}
Expand Down
24 changes: 21 additions & 3 deletions docs/adr/0014-workspace-checkpoints.md
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,23 @@ Registry changes hold ADR 0013's project `writer.lock` and then
flush, atomic replace, and directory flush. No other lock order is permitted.
An interruption exposes exactly the previous or next valid registry revision.

Acquisition is nonblocking (`try_lock_*`). Contention returns `GF_WRITER_BUSY`
with a precise message:

- `checkpoint mutation could not acquire writer.lock` when the writer is held;
- `checkpoint mutation could not acquire checkpoints.lock` when the writer was
acquired but `checkpoints.lock` is held (shared reader, exclusive mutator, or
a same-project window that released the writer first);
- `checkpoint read could not acquire checkpoints.lock` for shared read contention.

Release order is the reverse of acquisition: unlock `checkpoints.lock`, then
`writer.lock`. Recovery and mutation guards must not drop the writer while still
holding `checkpoints.lock`, so concurrent mutators never observe "writer free,
checkpoints busy" solely because of recovery teardown. Revert may still transfer
the writer into publication while retaining `checkpoints.lock` until the revert
returns; that window is intentional same-project contention and surfaces the
checkpoints-specific `GF_WRITER_BUSY` message.

### Creation, deletion, and idempotency

`checkpoint` resolves and leases `CURRENT`, validates the complete selected
Expand Down Expand Up @@ -206,9 +223,10 @@ actor_uuid_bytes_if_present)`, using the same exact presence-byte rules.

Mutation and recovery lock order is always project `writer.lock`, then
`checkpoints.lock`, then post-lock generation resolution/lease/validation.
The selected generation lease is held through registry publication. Deletion
does not acquire the generation lease; any independent open lease continues to
protect its generation from cleanup.
Release is always `checkpoints.lock` then `writer.lock`. The selected
generation lease is held through registry publication. Deletion does not
acquire the generation lease; any independent open lease continues to protect
its generation from cleanup.

### Retention, leases, and recovery

Expand Down