amadan

carlos / platform Public

instances import: an interrupt leaves a live import with its token and the commands to finish it (Plan 6 Task 6 fix round 1)

The flow runs under carlos db's signal context (Ctrl-C, SIGTERM, SIGHUP,
SIGPIPE). A cancellation after begin ends through the live exit: SD-N72 if
not yet shown, then SD-N73, exit 1, with no commit or abort. A cancelled
verify is not a verdict. Also: --verify is parsed before begin, a hand
commit ends the check and swap waits with SD-N75, and the upload accepts
only the door's 202.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0fc95020badca500aea1cb4726b48dadcfb31ff1 parent 87a1ead
3 files changed, +177 −7
  • apimode.go +1 −1
  • instances_import.go +43 −5
  • instances_import_test.go +133 −1
diff --git a/apimode.go b/apimode.go
index 01c52e3ff..1c3e99884 100644
--- a/apimode.go
+++ b/apimode.go
@@ -3777,7 +3777,7 @@ func (s *apiSession) importUpload(ctx context.Context, a apiAcct, app, host stri
}
defer res.Body.Close()
raw, _ := io.ReadAll(io.LimitReader(res.Body, 1<<20))
- if res.StatusCode != http.StatusAccepted && res.StatusCode != http.StatusOK {
+ if res.StatusCode != http.StatusAccepted {
return apiErrorFromBody(res.StatusCode, raw)
}
return nil
diff --git a/instances_import.go b/instances_import.go
index 8b8ed7690..aa2bba149 100644
--- a/instances_import.go
+++ b/instances_import.go
@@ -10,6 +10,7 @@ import (
"io"
"log"
"net/http"
+ "net/url"
"os"
"slices"
"strconv"
@@ -79,6 +80,14 @@ func runInstancesImport(args []string) {
fs.Usage()
os.Exit(2)
}
+ // A --verify that cannot make a request would fail only after the swap,
+ // and abort a good import over a typo.
+ if *verify != "" {
+ if _, err := url.Parse(importVerifyTarget(strings.TrimSpace(*host), strings.TrimSpace(*verify))); err != nil {
+ fs.Usage()
+ os.Exit(2)
+ }
+ }
sess := requireAPISession("import", *consoleFlag)
printUsing(stdout, sess)
acct := resolvedAccount(sess, acctRaw, acctIDRaw)
@@ -87,10 +96,17 @@ func runInstancesImport(args []string) {
if merr != nil {
log.Fatalf("instances import: %v", merr)
}
- err := sess.runImport(context.Background(), stdout, importJob{acct: acct, app: app2, host: strings.TrimSpace(*host), file: *file,
+ // Ctrl-C, SIGTERM or a closed terminal cancels the import's context
+ // rather than killing the process, so an import left live still shows
+ // its token and the commands to finish it (M17). The signals are
+ // carlos db's.
+ ctx, stop := dbSignalContext()
+ defer stop()
+ err := sess.runImport(ctx, stdout, importJob{acct: acct, app: app2, host: strings.TrimSpace(*host), file: *file,
replaces: strings.TrimSpace(*replaces), environment: strings.TrimSpace(*environment), verify: strings.TrimSpace(*verify), message: message})
switch {
case errors.Is(err, errImportFailed):
+ stop()
os.Exit(1)
case err != nil:
log.Fatalf("instances import: %v", err)
@@ -100,7 +116,10 @@ func runInstancesImport(args []string) {
// runImport hashes the file and begins, then drives the import (driveImport)
// and settles what its ending prints. The verify token stays in memory. It
// is printed (SD-N72) right after begin without --verify, and with --verify
-// only before an exit that leaves the import live and uncommitted.
+// only before an exit that leaves the import live and uncommitted. A
+// cancelled ctx after begin is such an exit: the token and the commands to
+// finish by hand, and no commit or abort is sent. Cancelled before begin
+// answers, it is begin's error.
func (s *apiSession) runImport(ctx context.Context, w io.Writer, j importJob) error {
f, err := os.Open(j.file)
if err != nil {
@@ -142,6 +161,10 @@ func (s *apiSession) runImport(ctx context.Context, w io.Writer, j importJob) er
return errImportFailed
}
showToken()
+ if ctx.Err() != nil {
+ fmt.Fprintln(w, importFinishLine(j.app, j.host))
+ return errImportFailed
+ }
if errors.As(err, &live) {
for _, l := range live.lines {
fmt.Fprintln(w, l)
@@ -163,7 +186,7 @@ func (s *apiSession) driveImport(ctx context.Context, w io.Writer, j importJob,
}
// The console answered 202 and checks the file on its own; the record
// stays awaiting-upload until it is kept.
- if _, err := s.pollSeed(ctx, j.acct, j.app, j.host, importCheckWait, importLook(id, "uploaded", "ready", "swapped", "placed")); err != nil {
+ if _, err := s.pollSeed(ctx, j.acct, j.app, j.host, importCheckWait, importLook(id, "uploaded", "ready", "swapped", "placed", "committed")); err != nil {
return err
}
if err := s.importStep(ctx, j.acct, j.app, j.host, id, "stopped"); err != nil {
@@ -171,13 +194,19 @@ func (s *apiSession) driveImport(ctx context.Context, w io.Writer, j importJob,
}
fmt.Fprintln(w, fmt.Sprintf("uploaded. The old process of %s is marked stopped, and the box swaps it on its next pass.", j.host))
fmt.Fprintln(w, fmt.Sprintf("waiting for the box to swap %s", j.host))
- if _, err := s.pollSeed(ctx, j.acct, j.app, j.host, importSwapWait, importLook(id, "swapped", "placed")); err != nil {
+ swapped, err := s.pollSeed(ctx, j.acct, j.app, j.host, importSwapWait, importLook(id, "swapped", "placed", "committed"))
+ if err != nil {
if errors.Is(err, errSeedWaitOver) {
return importLive{lines: []string{fmt.Sprintf("the box has not swapped %s after %d minutes. Check carlos instances list, then commit or abort.",
j.host, int(importSwapWait/time.Minute))}}
}
return err
}
+ if swapped.Seed.State == "committed" {
+ // Committed by hand (the token was shown at begin) between two looks.
+ fmt.Fprintln(w, importCommittedLine(j.host))
+ return nil
+ }
fmt.Fprintln(w, fmt.Sprintf("swapped. Until it is committed, %s answers only requests that carry the verify token.", j.host))
if j.verify == "" {
fmt.Fprintln(w, importFinishLine(j.app, j.host))
@@ -188,6 +217,10 @@ func (s *apiSession) driveImport(ctx context.Context, w io.Writer, j importJob,
client = &http.Client{Timeout: 2 * time.Minute}
}
code, verr := verifyImport(ctx, client, j.host, j.verify, token)
+ if ctx.Err() != nil {
+ // Cancelled is not a verdict: neither commit nor abort.
+ return ctx.Err()
+ }
if verr == nil && code >= 200 && code < 300 {
fmt.Fprintln(w, fmt.Sprintf("verify passed (%d). Committing.", code))
fmt.Fprintln(w, "waiting for the box to finish placing the snapshot")
@@ -324,10 +357,15 @@ func (s *apiSession) runImportVerb(ctx context.Context, w io.Writer, a apiAcct,
// verifyImport is the verify request through the edge (spec §5.5) to
// https://<host><path>.
func verifyImport(ctx context.Context, client *http.Client, host, path, token string) (int, error) {
+ return verifyImportURL(ctx, client, importVerifyTarget(host, path), token)
+}
+
+// importVerifyTarget is the verify URL, https://<host><path>.
+func importVerifyTarget(host, path string) string {
if !strings.HasPrefix(path, "/") {
path = "/" + path
}
- return verifyImportURL(ctx, client, "https://"+host+path, token)
+ return "https://" + host + path
}
// verifyImportURL asks target with the verify token and returns the
diff --git a/instances_import_test.go b/instances_import_test.go
index c10be46e5..2800ce51b 100644
--- a/instances_import_test.go
+++ b/instances_import_test.go
@@ -9,6 +9,7 @@ import (
"io"
"net/http"
"net/http/httptest"
+ "net/url"
"os"
"path/filepath"
"strings"
@@ -293,6 +294,7 @@ func TestImportUploadRefusal(t *testing.T) {
// answer as configured, and the record walks through gets, one per look
// (the last one repeats).
type importConsole struct {
+ uploadHold func(r *http.Request) // runs before the upload answers, outside mu
mu sync.Mutex
calls []string
beginStatus int
@@ -310,6 +312,10 @@ func newImportConsole(gets ...func(w http.ResponseWriter)) *importConsole {
}
func (c *importConsole) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodPut && c.uploadHold != nil {
+ io.Copy(io.Discard, r.Body)
+ c.uploadHold(r)
+ }
c.mu.Lock()
defer c.mu.Unlock()
io.Copy(io.Discard, r.Body)
@@ -378,6 +384,11 @@ func importRecordID(id, state, reason string) func(w http.ResponseWriter) {
// (when not empty) answered by verifyStatus through a TLS server standing
// in for the instance's host. It returns the output and the error.
func importRun(t *testing.T, c *importConsole, verify string, verifyStatus ...int) (string, error) {
+ t.Helper()
+ return importRunCtx(t, context.Background(), c, verify, verifyStatus...)
+}
+
+func importRunCtx(t *testing.T, ctx context.Context, c *importConsole, verify string, verifyStatus ...int) (string, error) {
t.Helper()
ts := httptest.NewServer(c)
defer ts.Close()
@@ -403,7 +414,7 @@ func importRun(t *testing.T, c *importConsole, verify string, verifyStatus ...in
j.host, j.verifyClient = strings.TrimPrefix(vs.URL, "https://"), vs.Client()
}
var out bytes.Buffer
- err := newTestSession(ts.URL).runImport(context.Background(), &out, j)
+ err := newTestSession(ts.URL).runImport(ctx, &out, j)
// The output names the file and the host as the tests know them.
return strings.NewReplacer(j.file, "main.db", j.host, "main.shop.example.com").Replace(out.String()), err
}
@@ -731,3 +742,124 @@ func TestPollSeedLimitAndFailures(t *testing.T) {
t.Fatalf("err = %v, looks = %d", err, looks)
}
}
+
+// cancelOnWait makes the seedWait seam cancel ctx on the first wait that
+// matches, as Ctrl-C would mid-wait; earlier waits pass.
+func cancelOnWait(t *testing.T, cancel context.CancelFunc, match func(n int, d time.Duration) bool) {
+ t.Helper()
+ oldWait, oldEvery := seedWait, seedPollEvery
+ n := 0
+ seedWait = func(ctx context.Context, d time.Duration) error {
+ if n++; match(n, d) {
+ cancel()
+ }
+ return ctx.Err()
+ }
+ seedPollEvery = 2 * time.Second
+ t.Cleanup(func() { seedWait, seedPollEvery = oldWait, oldEvery })
+}
+
+// wantCancelledLive is an interrupted import that is left live: the token
+// and the commands to finish by hand, no commit or abort, exit non-zero.
+func wantCancelledLive(t *testing.T, c *importConsole, out string, err error) {
+ t.Helper()
+ if !errors.Is(err, errImportFailed) {
+ t.Fatalf("err = %v", err)
+ }
+ // Without --verify the token was shown at begin; with it, just before
+ // the commands. Either way once, and the commands last.
+ if !strings.HasSuffix(out, "\n"+importCommitLine+"\n") || strings.Count(out, importTokenLine) != 1 ||
+ strings.Contains(out, "verify failed") {
+ t.Fatalf("output:\n%s", out)
+ }
+ if c.saw("commit") || c.saw("abort") {
+ t.Fatalf("calls = %v", c.called())
+ }
+}
+
+func TestImportInterruptedDuringTheSwapWait(t *testing.T) {
+ for _, verify := range []string{"", "/up"} {
+ t.Run("verify="+verify, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ cancelOnWait(t, cancel, func(n int, _ time.Duration) bool { return n == 3 }) // the upload's look, then two swap looks
+ c := newImportConsole(importRecord("uploaded", ""), importRecord("ready", ""))
+ out, err := importRunCtx(t, ctx, c, verify, http.StatusOK)
+ wantCancelledLive(t, c, out, err)
+ if !strings.Contains(out, "waiting for the box to swap main.shop.example.com\n") {
+ t.Fatalf("output:\n%s", out)
+ }
+ })
+ }
+}
+
+// A verify cut short is not a verdict: no "verify failed", no abort.
+func TestImportInterruptedDuringTheVerifyRetries(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ cancelOnWait(t, cancel, func(_ int, d time.Duration) bool { return d == importVerifyRetry })
+ c := newImportConsole(importRecord("uploaded", ""), importRecord("swapped", ""))
+ out, err := importRunCtx(t, ctx, c, "/up", http.StatusServiceUnavailable)
+ wantCancelledLive(t, c, out, err)
+}
+
+func TestImportInterruptedDuringTheUpload(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ fakeSeedWait(t, 2*time.Second)
+ c := newImportConsole(importRecord("uploaded", ""))
+ c.uploadHold = func(r *http.Request) {
+ cancel()
+ <-r.Context().Done() // the client gives up the request
+ }
+ out, err := importRunCtx(t, ctx, c, "/up", http.StatusOK)
+ wantCancelledLive(t, c, out, err)
+ if c.saw("stopped") || c.saw("get") {
+ t.Fatalf("calls = %v", c.called())
+ }
+}
+
+func TestImportCancelledBeforeBeginJustExits(t *testing.T) {
+ fakeSeedWait(t, 2*time.Second)
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ c := newImportConsole()
+ out, err := importRunCtx(t, ctx, c, "/up", http.StatusOK)
+ if !errors.Is(err, context.Canceled) || errors.Is(err, errImportFailed) {
+ t.Fatalf("err = %v", err)
+ }
+ wantLines(t, out, "uploading main.db (3B)")
+ if len(c.called()) != 0 {
+ t.Fatalf("calls = %v", c.called())
+ }
+}
+
+// An import committed by hand between two looks ends the swap wait with
+// SD-N75 rather than running to SD-N79 (review minor 3).
+func TestImportCommittedByHandDuringTheSwapWait(t *testing.T) {
+ fakeSeedWait(t, 2*time.Second)
+ c := newImportConsole(importRecord("uploaded", ""), importRecord("committed", ""))
+ out, err := importRun(t, c, "")
+ if err != nil || !strings.HasSuffix(out, "waiting for the box to swap main.shop.example.com\ncommitted. main.shop.example.com is served by the platform now.\n") {
+ t.Fatalf("err = %v, output:\n%s", err, out)
+ }
+}
+
+func TestImportVerifyTarget(t *testing.T) {
+ for _, c := range []struct{ path, want string }{
+ {"/up", "https://main.shop.example.com/up"},
+ {"up", "https://main.shop.example.com/up"},
+ } {
+ if got := importVerifyTarget("main.shop.example.com", c.path); got != c.want {
+ t.Errorf("importVerifyTarget(%q) = %q, want %q", c.path, got, c.want)
+ }
+ }
+ // The CLI refuses a --verify that cannot become a request before it
+ // begins (review minor 2), with the parse the request itself uses.
+ if _, err := url.Parse(importVerifyTarget("main.shop.example.com", "/%zz")); err == nil {
+ t.Fatal("a path with a bad escape parsed")
+ }
+ if _, err := http.NewRequest(http.MethodGet, importVerifyTarget("main.shop.example.com", "/%zz"), nil); err == nil {
+ t.Fatal("the request would have been made")
+ }
+}