amadan

carlos / platform Public

carlos db: honour a stop after the downloads finish

Once every file was downloaded, Ctrl-C or SIGTERM during the local
compaction was ignored: os.File reads and litestream's compactor check
no context, so pull could still publish the copy and report success.

- the downloaded files check the restore's context on every read;
- RestoreTo checks it again before CheckDB, and reports a stop as
  context.Canceled rather than the compactor's unwrapped read error;
- copyOut checks it before creating --out, on every read of the copy,
  and once more after the fsync, removing --out if a stop landed, so
  success is printed only for a copy that was not interrupted.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d7a0c91279af60fadb1caa6609fbf8e4388bfbb7 parent 838ea5c
4 files changed, +94 −8
  • db.go +30 −4
  • db_test.go +19 −4
  • internal/dbdoor/restore.go +19 −0
  • internal/dbdoor/restore_test.go +26 −0

CI passed — run details

diff --git a/db.go b/db.go
index 0456dffc1..86d3df454 100644
--- a/db.go
+++ b/db.go
@@ -215,7 +215,7 @@ func dbPull(ctx context.Context, t dbTarget, out string) error {
return err
}
defer os.RemoveAll(r.Dir)
- if err := copyOut(r.Path, out); err != nil {
+ if err := copyOut(ctx, r.Path, out); err != nil {
return err
}
fmt.Fprintf(stdout, "Saved a copy of %s's database to %s.\n", t.host, out)
@@ -228,7 +228,11 @@ func dbPull(ctx context.Context, t dbTarget, out string) error {
// out. A copy and never os.Rename: the temp directory and the user's
// destination are routinely on different filesystems, where a rename
// fails. A failed copy leaves nothing behind at out.
-func copyOut(src, out string) (err error) {
+func copyOut(ctx context.Context, src, out string) (err error) {
+ // Nothing is published once the person has asked to stop.
+ if err := ctx.Err(); err != nil {
+ return err
+ }
in, err := os.Open(src)
if err != nil {
return err
@@ -252,13 +256,35 @@ func copyOut(src, out string) (err error) {
if err = f.Chmod(0o600); err != nil {
return err
}
- if _, err = io.Copy(f, in); err != nil {
+ if _, err = io.Copy(f, ctxReader{ctx, in}); err != nil {
return err
}
if err = f.Sync(); err != nil {
return err
}
- return f.Close()
+ if err = f.Close(); err != nil {
+ return err
+ }
+ // A stop that lands as the copy finishes still withdraws it: the
+ // caller reports success only for a copy that was not interrupted.
+ if err = ctx.Err(); err != nil {
+ os.Remove(out)
+ return err
+ }
+ return nil
+}
+
+// ctxReader stops a copy at the next read once ctx is done.
+type ctxReader struct {
+ ctx context.Context
+ r io.Reader
+}
+
+func (c ctxReader) Read(p []byte) (int, error) {
+ if err := c.ctx.Err(); err != nil {
+ return 0, err
+ }
+ return c.r.Read(p)
}
func runDBQuery(args []string) {
diff --git a/db_test.go b/db_test.go
index 243eab0ce..b7a65f520 100644
--- a/db_test.go
+++ b/db_test.go
@@ -127,14 +127,14 @@ func TestPullRefusesAnExistingOutAndLands0600(t *testing.T) {
if err := os.WriteFile(existing, []byte("mine"), 0o644); err != nil {
t.Fatal(err)
}
- if err := copyOut(src, existing); err == nil {
+ if err := copyOut(context.Background(), src, existing); err == nil {
t.Fatal("copyOut over an existing file succeeded")
}
if b, _ := os.ReadFile(existing); string(b) != "mine" {
t.Fatal("copyOut touched the existing file")
}
out := filepath.Join(dir, "new.db")
- if err := copyOut(src, out); err != nil {
+ if err := copyOut(context.Background(), src, out); err != nil {
t.Fatal(err)
}
fi, err := os.Stat(out)
@@ -158,7 +158,7 @@ func TestPullCopiesAcrossFilesystems(t *testing.T) {
if err := os.MkdirAll(filepath.Dir(out), 0o755); err != nil {
t.Fatal(err)
}
- if err := copyOut(src, out); err != nil {
+ if err := copyOut(context.Background(), src, out); err != nil {
t.Fatal(err)
}
if _, err := os.Stat(src); err != nil {
@@ -169,7 +169,7 @@ func TestPullCopiesAcrossFilesystems(t *testing.T) {
// A failed copy leaves nothing at --out.
func TestPullLeavesNoPartialFile(t *testing.T) {
out := filepath.Join(t.TempDir(), "pulled.db")
- if err := copyOut(filepath.Join(t.TempDir(), "missing.db"), out); err == nil {
+ if err := copyOut(context.Background(), filepath.Join(t.TempDir(), "missing.db"), out); err == nil {
t.Fatal("copyOut of a missing source succeeded")
}
if _, err := os.Stat(out); !os.IsNotExist(err) {
@@ -493,3 +493,18 @@ func TestQueryRefusesDisguisedPragmaChanges(t *testing.T) {
}
}
}
+
+// A pull interrupted before or during the copy publishes nothing at --out,
+// and so cannot go on to report success.
+func TestPullPublishesNothingOnceCancelled(t *testing.T) {
+ src := fixtureDB(t)
+ out := filepath.Join(t.TempDir(), "pulled.db")
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ if err := copyOut(ctx, src, out); !errors.Is(err, context.Canceled) {
+ t.Fatalf("copyOut after cancel = %v, want context.Canceled", err)
+ }
+ if _, err := os.Stat(out); !os.IsNotExist(err) {
+ t.Fatalf("a cancelled pull left %s (%v)", out, err)
+ }
+}
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
index dd47e4621..859d0f43f 100644
--- a/internal/dbdoor/restore.go
+++ b/internal/dbdoor/restore.go
@@ -85,6 +85,11 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
continue
}
os.RemoveAll(dir)
+ // A stop reaches litestream's local half as a broken read, which
+ // its compactor reports without wrapping. Name the real cause.
+ if cerr := ctx.Err(); cerr != nil {
+ return nil, cerr
+ }
switch {
case errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable):
return nil, ErrNothingReplicated
@@ -93,6 +98,10 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
}
return nil, err
}
+ if err := ctx.Err(); err != nil {
+ os.RemoveAll(dir)
+ return nil, err
+ }
if err := checkDB(out); err != nil {
os.RemoveAll(dir)
return nil, fmt.Errorf("%w: %v", ErrCheckFailed, err)
@@ -225,6 +234,16 @@ type downloaded struct {
*os.File
}
+// Read checks the restore's context first. Once every file is downloaded,
+// litestream's compaction reads only these files and checks no context of
+// its own, so this is where Ctrl-C reaches it.
+func (d *downloaded) Read(p []byte) (int, error) {
+ if err := d.ctx.Err(); err != nil {
+ return 0, err
+ }
+ return d.File.Read(p)
+}
+
func (d *downloaded) Close() error {
err := d.File.Close()
if rerr := os.Remove(d.File.Name()); err == nil {
diff --git a/internal/dbdoor/restore_test.go b/internal/dbdoor/restore_test.go
index e169f1e14..6cad1ae8a 100644
--- a/internal/dbdoor/restore_test.go
+++ b/internal/dbdoor/restore_test.go
@@ -271,3 +271,29 @@ func TestRestoreToGivesUpOnALinkThatMakesNoProgress(t *testing.T) {
}
assertEmptyDir(t, parent)
}
+
+// Ctrl-C after the last file has arrived still stops the restore. The
+// local half (litestream's compaction over the downloaded files) does not
+// check the context itself, so the downloaded files do, and RestoreTo
+// checks again before CheckDB. Nothing is handed over, nothing left.
+func TestRestoreToStopsWhenCancelledAfterTheDownloads(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 3)
+ door := newTestDoor(t, src)
+ files := len(collect(t, NewClient(door.do), 0, 0))
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ var gets atomic.Int32
+ do := func(c context.Context, tail string, h http.Header) (*http.Response, error) {
+ res, err := door.do(c, tail, h)
+ if err == nil && strings.Contains(strings.TrimPrefix(tail, "/ltx/"), "/") && int(gets.Add(1)) == files {
+ cancel() // the last download is in hand; only local work is left
+ }
+ return res, err
+ }
+ parent := t.TempDir()
+ if _, err := RestoreTo(ctx, NewClient(do), parent); !errors.Is(err, context.Canceled) {
+ t.Fatalf("RestoreTo cancelled after its downloads = %v, want context.Canceled", err)
+ }
+ assertEmptyDir(t, parent)
+}