bottomless: fix snapshot upload pipeline silently dropping generation snapshots

The continuous-backup pipeline could run for days while every generation it
created was missing its base snapshot in object storage, degrading or breaking
restore, and in one failure mode blocking checkpoints entirely so the WAL grew
without bound. Five interacting defects:

- A failed snapshot upload was never retried and blocked checkpoints until a
  process restart: the snapshot task attempted its PutObject exactly once and
  nothing ever re-ran it for the current generation, so the checkpoint gate
  refused every subsequent checkpoint with SQLITE_BUSY. The upload is now
  retried with backoff inside the snapshot task (rebuilding the body stream
  per attempt, so a retry never sends a stale Content-Length), and the
  checkpoint gate re-triggers the snapshot of the current generation when it
  failed or never ran. Re-uploading at any later point is correct because the
  main database file only changes on a TRUNCATE checkpoint, and checkpoints
  stay refused until the upload succeeds.

- A checkpoint attempt that found an empty WAL unconditionally marked the
  current generation as snapshotted, laundering failed or still-in-flight
  uploads into success markers and letting the next checkpoint rotate the
  generation while leaving no snapshot of it behind. The empty-WAL branch now
  re-triggers a missing snapshot instead of overwriting the failure state.

- Concurrent snapshot tasks compressed the database into the same fixed
  db.gz/db.zstd path with truncate(true), shrinking the file under the other
  task's in-flight upload stream (surfacing as hyper BodyWriteAborted/NotEof
  dispatch failures) or silently uploading a torn object. Compression now
  targets a per-generation artifact path, an in-flight flag guarantees at most
  one snapshot task at a time, and stale artifacts (including legacy
  fixed-name ones) are swept only while no upload is running. S3 object keys
  are unchanged.

- The AWS SDK's stalled-stream protection aborted slow snapshot uploads: a
  whole-database PutObject can run for minutes and legitimately stalls beyond
  the ~5s grace period whenever the process is busy, and the watchdog abort
  surfaces as the same BodyWriteAborted signature. The watchdog is now
  disabled on the bottomless client; bottomless owns its retry policy.

- Restarting on top of a non-empty database file created a generation with no
  .dep parent link, severing the chain that restore walks when a generation
  has no snapshot: compare_with_local returned early before storing the
  generation it compared against. The store now happens before the early
  return.

Snapshot upload log lines now include the database name and generation.

Tests: new snapshot_pipeline regression tests in libsql-server drive the
bottomless WAL wrapper against a mock S3 server with failure injection,
covering upload retry, checkpoint recovery without restart (including the
empty-WAL laundering case), the snapshot overlap guard, and the .dep link on
restart. Existing bottomless tests pass, including the ignored backup_restore
test.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013Q5tV6oA1zXfrWhchz2Cqk

Author: claude

.gitignore

@@ -5,6 +5,10 @@
 
 libsql-server/testrollbackrestore/**
 libsql-server/testbackuprestore/**
+libsql-server/testsnapshotretry/**
+libsql-server/testsnapshotwedge/**
+libsql-server/testsnapshotoverlap/**
+libsql-server/testsnapshotdep/**
 libsql-sqlite3/**.o.tmp
 /libsql-ffi/bundled/SQLite3MultipleCiphers/build/**
 # will be copied from bundled/src

bottomless/src/bottomless_wal.rs

@@ -32,6 +32,28 @@ impl BottomlessWalWrapper {
     }
 }
 
+/// Returns true if the current generation's snapshot upload has completed.
+/// When the upload failed, or was never started (e.g. it was interrupted by a
+/// process restart or a crashed task), it is re-triggered so that the backup
+/// pipeline heals itself instead of refusing checkpoints until the process is
+/// restarted. Re-uploading the snapshot of the current generation at any later
+/// point in time is correct: the main database file only changes on a TRUNCATE
+/// checkpoint, and checkpoints are refused until the snapshot upload succeeds.
+async fn ensure_snapshotted(replicator: &mut Replicator) -> bool {
+    if replicator.is_snapshotted().await {
+        return true;
+    }
+    if replicator.snapshot_in_flight() {
+        tracing::debug!("snapshot upload for the current generation is still in flight");
+    } else {
+        tracing::warn!("current generation is missing its snapshot, re-triggering snapshot");
+        if let Err(e) = replicator.snapshot_main_db_file(false).await {
+            tracing::error!("failed to re-trigger a snapshot upload: {e}");
+        }
+    }
+    false
+}
+
 impl<T: Wal> WrapWal<T> for BottomlessWalWrapper {
     fn savepoint_undo(
         &mut self,
@@ -136,7 +158,14 @@ impl<T: Wal> WrapWal<T> for BottomlessWalWrapper {
                         tracing::debug!(
                             "No committed changes in this generation, not snapshotting"
                         );
-                        replicator.skip_snapshot_for_current_generation();
+                        // The checkpoint is skipped, but the current generation
+                        // still needs its snapshot for the next checkpoint to be
+                        // allowed, so re-trigger it here if it failed. The failed
+                        // state must never be overwritten with a success marker:
+                        // that would let the next checkpoint rotate the generation
+                        // while leaving no snapshot of it behind, silently
+                        // degrading restore.
+                        ensure_snapshotted(replicator).await;
                         return Err(Error::new(SQLITE_BUSY));
                     }
 
@@ -163,8 +192,7 @@ impl<T: Wal> WrapWal<T> for BottomlessWalWrapper {
                         }
                     }
                     tracing::debug!("commited after {:?}", before.elapsed());
-                    let snapshotted = replicator.is_snapshotted().await;
-                    if !snapshotted {
+                    if !ensure_snapshotted(replicator).await {
                         tracing::warn!("previous generation not snapshotted, skipping checkpoint");
                         return Err(Error::new(SQLITE_BUSY));
                     }