amadan

carlos / platform Public

console: the import upload, spooled and hashed, then checked and put detached (Plan 6 Task 2)

PUT .../instances/{host}/import spools the body to a private from-* dir
under MaxBytesReader(bytes+1), hashing as it writes; a length or hash
mismatch answers 422 and leaves the import awaiting-upload. Once the spool
matches it answers 202 {"state":"checking"} and runs the integrity check,
the create-only put and the move to uploaded detached, as runFrom does
(preflight M6, a deviation from spec §5.2 for Paul); a failure aborts the
seed with the deck sentence as its reason. importGate refuses on a claim
console before it reads the record. Wires SD-N51-53, 59, 61, 63.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 88c39aca86709d62234565495855527967024b46 parent 66054e5
10 files changed, +692 −4
  • approved_seeded_copy_test.go +3 −2
  • internal/console/apiinstanceimport.go +284 −0
  • internal/console/apiinstanceimport_test.go +381 −0
  • internal/console/authcli.go +3 −0
  • internal/console/claim_invariant_fixtures_test.go +9 −0
  • internal/console/claim_invariant_test.go +2 −0
  • internal/console/server.go +3 −2
  • internal/console/svcroutes_test.go +5 −0
  • internal/console/testdata/claim-authority-sites.txt +1 −0
  • internal/console/testdata/claim-routes.txt +1 −0
diff --git a/approved_seeded_copy_test.go b/approved_seeded_copy_test.go
index 05b0ab82a..911a8b00f 100644
--- a/approved_seeded_copy_test.go
+++ b/approved_seeded_copy_test.go
@@ -35,11 +35,12 @@ var seededCopyIDs = func() []string {
// Task 3 wires SD-N1, Task 5 SD-N2 to SD-N8 and SD-N113, Task 6 SD-N9 to SD-N13,
// SD-N111 and SD-N112; Plan 4 Task 3 SD-N14 to SD-N19; Plan 5 Task 3 SD-N27 to SD-N29,
// Task 5 SD-N20 to SD-N26, SD-N30 to SD-N39 and SD-N41; Plan 5 Task 6 SD-N40, SD-N42 to SD-N47, SD-N83 to SD-N88 and SD-N114;
-// Plan 6 Task 1 SD-N48 to SD-N50, SD-N62 and SD-N67.
+// Plan 6 Task 1 SD-N48 to SD-N50, SD-N62 and SD-N67; Plan 6 Task 2 SD-N51 to SD-N53, SD-N59, SD-N61 and SD-N63.
var seededWiredIDs = []string{"SD-N1", "SD-N2", "SD-N3", "SD-N4", "SD-N5", "SD-N6", "SD-N7", "SD-N8", "SD-N113", "SD-N9", "SD-N10", "SD-N11", "SD-N12", "SD-N13", "SD-N111", "SD-N112", "SD-N14", "SD-N15", "SD-N16", "SD-N17", "SD-N18", "SD-N19", "SD-N27", "SD-N28", "SD-N29",
"SD-N20", "SD-N21", "SD-N22", "SD-N23", "SD-N24", "SD-N25", "SD-N26", "SD-N30", "SD-N31", "SD-N32", "SD-N33", "SD-N34", "SD-N35", "SD-N36", "SD-N37", "SD-N38", "SD-N39", "SD-N41",
"SD-N40", "SD-N42", "SD-N43", "SD-N44", "SD-N45", "SD-N46", "SD-N47", "SD-N114", "SD-N83", "SD-N84", "SD-N85", "SD-N86", "SD-N87", "SD-N88",
- "SD-N48", "SD-N49", "SD-N50", "SD-N62", "SD-N67"}
+ "SD-N48", "SD-N49", "SD-N50", "SD-N62", "SD-N67",
+ "SD-N51", "SD-N52", "SD-N53", "SD-N59", "SD-N61", "SD-N63"}
func mustParseSeededRecord(t *testing.T) []copyRow {
t.Helper()
diff --git a/internal/console/apiinstanceimport.go b/internal/console/apiinstanceimport.go
index 41f8f3d1d..e8ea428ec 100644
--- a/internal/console/apiinstanceimport.go
+++ b/internal/console/apiinstanceimport.go
@@ -9,16 +9,22 @@ package console
import (
"context"
+ "crypto/sha256"
+ "encoding/hex"
"errors"
"fmt"
+ "io"
"log"
"net"
"net/http"
"os"
+ "path/filepath"
"regexp"
+ "runtime/debug"
"strconv"
"time"
+ "github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/seeds"
@@ -239,3 +245,281 @@ func (s *Server) unbeginImport(ctx context.Context, c seedCreate, id, why string
log.Printf("console: import %s: finish seed %s after %s: %v", host, id, why, err)
}
}
+
+// Seams for the tests.
+var (
+ // checkSeedDBConsole is the upload's integrity check (spec §5.2 step 3).
+ checkSeedDBConsole = lease.CheckDBContext
+ // importCheckDeadline bounds that check.
+ importCheckDeadline = 30 * time.Minute
+ // importSpoolHook, when set, sees the spool once the body is read and
+ // before it is checked or removed.
+ importSpoolHook func(path string)
+)
+
+// importUploadReadDeadline is how long the upload's body may take to
+// arrive: up to 4 GiB, past the server's ReadTimeout. It covers the body
+// read only; the check and the put run after the answer.
+const importUploadReadDeadline = 60 * time.Minute
+
+// importGate is every import call's gate: the instance gate (cliRoute has
+// already applied the instance role), the claim console's refusal, the
+// host, the record, and that the record is an import (SD-N59).
+func (s *Server) importGate(w http.ResponseWriter, r *http.Request) (principal string, acct *Account, app, host string, rec *provision.Record, ok bool) {
+ principal, acct, app, ok = s.instanceGate(w, r)
+ if !ok {
+ return "", nil, "", "", nil, false
+ }
+ // A claim console begins no import (seedRegionReady), and its claim
+ // protocol has no place for one yet (#250): every import door refuses
+ // there before it reads the record.
+ if s.cfg.HostClaim {
+ ref := seedClaimConsoleRefusal()
+ jsonError(w, ref.Status, ref.Msg)
+ return "", nil, "", "", nil, false
+ }
+ host = r.PathValue("host")
+ notInstance := fmt.Sprintf("%s is not an instance of %s", host, app)
+ if !provision.ValidHostShape(host) {
+ jsonError(w, http.StatusNotFound, notInstance)
+ return "", nil, "", "", nil, false
+ }
+ ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
+ defer cancel()
+ rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
+ if errors.Is(err, os.ErrNotExist) {
+ jsonError(w, http.StatusNotFound, notInstance)
+ return "", nil, "", "", nil, false
+ }
+ if err != nil {
+ log.Printf("console: import %s/%s/%s: read the record: %v", acct.Sqid, app, host, err)
+ jsonError(w, http.StatusBadGateway, importFailed)
+ return "", nil, "", "", nil, false
+ }
+ if !rec.Seed.Import() {
+ jsonError(w, http.StatusConflict, fmt.Sprintf("%s is not an import.", host))
+ return "", nil, "", "", nil, false
+ }
+ return principal, acct, app, host, rec, true
+}
+
+// importUpload is an upload whose spool matched: everything
+// runImportUpload needs once the request has been answered. It owns the
+// spool's directory and the scratch reservation from the hand-over.
+type importUpload struct {
+ ctx context.Context // the request's values, never its cancellation
+ acct *Account
+ app, host, by string
+ id string
+ dir, spool string
+ release func()
+}
+
+// handleAPIImportUpload is spec §5.2, in two halves (preflight M6, a
+// deviation from the spec's synchronous 422 and 504 for Paul, following
+// Plan 5 ruling R2). The request half reads the body into a private spool
+// and checks its length and sha256 against begin's: a mismatch answers 422
+// and leaves the import awaiting-upload, so the right file can be sent.
+// Once the spool matches it answers 202 {"state":"checking"} and the rest
+// runs detached (runImportUpload): an edge backhaul's 30-second
+// response-header timeout and idle NATs would cut a response that waited
+// for a 30-minute check and a 4 GiB put. The caller follows the record
+// until it is uploaded, or aborted with the reason in Seed.Reason.
+func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) {
+ principal, acct, app, host, rec, ok := s.importGate(w, r)
+ if !ok {
+ return
+ }
+ ctx := r.Context()
+ seed := rec.Seed
+ if seed.State != seeds.StateAwaitingUpload {
+ jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. It takes an upload only while it waits for one.", host, seed.State))
+ return
+ }
+ // The snapshot key is built from the record (preflight M8): an id that
+ // cannot name a key must never reach the put.
+ if err := seeds.Check(host, seed.ID); err != nil {
+ log.Printf("console: import %s: seed %q: %v", host, seed.ID, err)
+ jsonError(w, http.StatusBadGateway, importFailed)
+ return
+ }
+ if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" {
+ ref := seedNoDataRefusal()
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
+ checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
+ ref := s.seedRegionReady(checks, rec.Region, true)
+ cancel()
+ if ref != nil {
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
+ releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, seed.Bytes)
+ if ref != nil {
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
+ dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix)
+ if err != nil {
+ releaseScratch()
+ log.Printf("console: import %s: scratch: %v", host, err)
+ ref := seedScratchFull()
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
+ handedOver := false
+ defer func() {
+ if !handedOver {
+ _ = os.RemoveAll(dir) // with any -shm or -wal beside the spool
+ releaseScratch()
+ }
+ }()
+ spool := filepath.Join(dir, host+".db")
+ if !s.spoolImport(w, r, spool, seed) {
+ return
+ }
+ writeJSON(w, http.StatusAccepted, map[string]string{"state": "checking"})
+ handedOver = true
+ s.seedCopies.Add(1)
+ go s.runImportUpload(&importUpload{ctx: context.WithoutCancel(ctx), acct: acct, app: app, host: host, by: principal,
+ id: seed.ID, dir: dir, spool: spool, release: releaseScratch})
+}
+
+// spoolImport reads the body into spool, never more than the declared
+// length and one byte, hashing as it writes, and reports whether it
+// matches begin's length and sha256. It has answered w when it reports
+// false.
+func (s *Server) spoolImport(w http.ResponseWriter, r *http.Request, spool string, seed *provision.SeedSource) bool {
+ host := r.PathValue("host")
+ // The server's ReadTimeout would cut a large body; this read gets an
+ // hour. Go drops the deadline once the body has been read.
+ if err := http.NewResponseController(w).SetReadDeadline(time.Now().Add(importUploadReadDeadline)); err != nil {
+ log.Printf("console: import %s: extend the read deadline: %v", host, err)
+ }
+ f, err := os.OpenFile(spool, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
+ if err != nil {
+ log.Printf("console: import %s: spool: %v", host, err)
+ ref := seedScratchFull()
+ jsonError(w, ref.Status, ref.Msg)
+ return false
+ }
+ h := sha256.New()
+ n, cerr := io.Copy(io.MultiWriter(f, h), http.MaxBytesReader(w, r.Body, seed.Bytes+1))
+ ferr := f.Close()
+ if importSpoolHook != nil {
+ importSpoolHook(spool)
+ }
+ var tooLong *http.MaxBytesError
+ switch {
+ case cerr != nil && !errors.As(cerr, &tooLong):
+ // The caller went away, the body broke off, or the spool's disk
+ // failed: nothing to keep, and the import still waits.
+ log.Printf("console: import %s: read the upload: %v", host, cerr)
+ jsonError(w, http.StatusBadGateway, importFailed)
+ return false
+ case ferr != nil:
+ log.Printf("console: import %s: spool: %v", host, ferr)
+ jsonError(w, http.StatusBadGateway, importFailed)
+ return false
+ case tooLong != nil || n != seed.Bytes || hex.EncodeToString(h.Sum(nil)) != seed.SHA256:
+ jsonError(w, http.StatusUnprocessableEntity, "the upload does not match the length and sha256 this import began with, so it was not kept.")
+ return false
+ }
+ return true
+}
+
+// runImportUpload is the detached half: the integrity check, the put to
+// the seed's snapshot key, and the move to uploaded. It never depends on
+// the request, which has been answered. Every failure moves the record to
+// aborted with the sentence the door would have answered in Seed.Reason
+// (SD-N52, SD-N61 or SD-N67). On every exit it removes the spool's
+// directory and then releases the reservation, as runFrom does.
+//
+// The console has no shutdown hook to cancel a running check: a restart
+// leaves the record awaiting-upload, its spool is cleared at start, and
+// the sweep aborts it (Plan 8).
+func (s *Server) runImportUpload(j *importUpload) {
+ defer s.seedCopies.Done()
+ defer j.release()
+ ctx := j.ctx
+ sqid, app, host := j.acct.Sqid, j.app, j.host
+ abort := func(reason, why string) {
+ bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
+ defer cancel()
+ if _, err := provision.SetRecordSeed(bg, s.cfg.Releases, sqid, app, host, provision.SeedAbortBecause(j.id, reason)); err != nil {
+ log.Printf("console: import %s: abort seed %s after %s: %v", host, j.id, why, err)
+ }
+ s.bumpSerial(bg)
+ }
+ // A panic in the check or the put would end the whole console: it
+ // aborts this seed instead. Registered after release and Done, so it
+ // runs before them; the spool's removal, registered later, has
+ // already run.
+ defer func() {
+ if p := recover(); p != nil {
+ log.Printf("console: import %s: panic in the upload of seed %s: %v\n%s", host, j.id, p, debug.Stack())
+ abort(importFailed, "a panic")
+ }
+ }()
+ defer os.RemoveAll(j.dir) // with any -shm or -wal the check left
+
+ cctx, ccancel := context.WithTimeout(ctx, importCheckDeadline)
+ err := checkSeedDBConsole(cctx, j.spool)
+ timedOut := errors.Is(cctx.Err(), context.DeadlineExceeded)
+ ccancel()
+ if err != nil {
+ if timedOut {
+ abort("checking the upload took longer than 30 minutes, so it was not kept.", "the check's deadline")
+ return
+ }
+ log.Printf("console: import %s: check: %v", host, err)
+ abort("the upload is not a database that passes an integrity check, so it was not kept.", "the check")
+ return
+ }
+ uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline)
+ err = putSnapshot(uctx, s.cfg.SeedData, j.spool, seeds.SnapshotKey(host, j.id))
+ ucancel()
+ if err != nil {
+ log.Printf("console: import %s: put the snapshot: %v", host, err)
+ abort(importFailed, "the put")
+ return
+ }
+ pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout)
+ defer pcancel()
+ if _, err := provision.SetRecordSeed(pctx, s.cfg.Releases, sqid, app, host, provision.SeedUploaded(j.id)); err != nil {
+ log.Printf("console: import %s: record uploaded: %v", host, err)
+ abort(importFailed, "the record write")
+ return
+ }
+ s.auditActivity(j.acct, app, j.by, Activity{Action: "import-uploaded", TargetType: "instance", Target: host,
+ Detail: fmt.Sprintf("uploaded the snapshot for %s", host)})
+}
+
+// putSnapshot writes path to key with one create-only PutFrom (spec §5.2
+// step 4). An object already there is a failed earlier attempt for this
+// same seed (the key is scoped to its id): it is deleted and the put
+// retried once.
+func putSnapshot(ctx context.Context, data release.Store, path, key string) error {
+ put := func() error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ st, err := f.Stat()
+ if err != nil {
+ return err
+ }
+ _, err = release.PutObjectFrom(ctx, data, key, f, st.Size(), "")
+ return err
+ }
+ err := put()
+ if errors.Is(err, release.ErrCASConflict) {
+ if err := data.Delete(ctx, key); err != nil {
+ return err
+ }
+ err = put()
+ }
+ return err
+}
diff --git a/internal/console/apiinstanceimport_test.go b/internal/console/apiinstanceimport_test.go
index 48b75b86b..dffb09711 100644
--- a/internal/console/apiinstanceimport_test.go
+++ b/internal/console/apiinstanceimport_test.go
@@ -3,12 +3,17 @@ package console
import (
"bytes"
"context"
+ "crypto/sha256"
+ "database/sql"
+ "encoding/hex"
"encoding/json"
"log"
"net/http"
"os"
+ "path/filepath"
"strings"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -473,3 +478,379 @@ func TestImportBeginUploadsNothing(t *testing.T) {
f.requireNothingUploaded(t)
f.requireClean(t)
}
+
+// ---- Upload (spec §5.2, preflight M5 to M8) ----
+
+const (
+ importMismatch = "the upload does not match the length and sha256 this import began with, so it was not kept."
+ importNotADB = "the upload is not a database that passes an integrity check, so it was not kept."
+ importSlow = "checking the upload took longer than 30 minutes, so it was not kept."
+)
+
+// uploadImport puts body to host's import door and returns what it
+// answered.
+func uploadImport(t *testing.T, f fromFixture, host string, body []byte) (int, map[string]any) {
+ t.Helper()
+ req, err := http.NewRequest(http.MethodPut, f.url+"/"+host+"/import", bytes.NewReader(body))
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Authorization", "Bearer "+f.token)
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer res.Body.Close()
+ var out map[string]any
+ _ = json.NewDecoder(res.Body).Decode(&out)
+ return res.StatusCode, out
+}
+
+// sqliteBytes is a small, sound SQLite database and its sha256.
+func sqliteBytes(t *testing.T) ([]byte, string) {
+ t.Helper()
+ p := filepath.Join(t.TempDir(), "snap.db")
+ db, err := sql.Open("sqlite", p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Exec(`CREATE TABLE t (x INTEGER); INSERT INTO t VALUES (1)`); err != nil {
+ t.Fatal(err)
+ }
+ db.Close()
+ b, err := os.ReadFile(p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b, sha256Of(b)
+}
+
+func sha256Of(b []byte) string {
+ sum := sha256.Sum256(b)
+ return hex.EncodeToString(sum[:])
+}
+
+// begun begins an import of host for body and returns its record.
+func (f fromFixture) begun(t *testing.T, host string, body []byte) *provision.Record {
+ t.Helper()
+ code, out := beginImport(t, f, host, sha256Of(body), int64(len(body)))
+ rec, _ := f.requireBegun(t, code, out, host)
+ return rec
+}
+
+// uploaded puts body, requires the 202 that hands it to the check, and
+// waits for the detached half to finish.
+func (f fromFixture) uploaded(t *testing.T, host string, body []byte) *provision.Record {
+ t.Helper()
+ if code, out := uploadImport(t, f, host, body); code != http.StatusAccepted || len(out) != 1 || out["state"] != "checking" {
+ t.Fatalf("upload = %d %v, want 202 {\"state\":\"checking\"}", code, out)
+ }
+ f.fx.srv.seedCopies.Wait()
+ rec, err := f.record(t, host)
+ if err != nil || rec.Seed == nil {
+ t.Fatalf("record %s after its upload = %+v, %v", host, rec, err)
+ }
+ return rec
+}
+
+func requireImportState(t *testing.T, rec *provision.Record, id, state, reason string) {
+ t.Helper()
+ if rec.Seed == nil || rec.Seed.ID != id || rec.Seed.Kind != seeds.KindImport || rec.Seed.State != state || rec.Seed.Reason != reason {
+ t.Fatalf("record %s seed = %+v; want %s %s with reason %q", rec.Host, rec.Seed, id, state, reason)
+ }
+}
+
+func TestImportUploadStoresTheSnapshot(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ rec := f.uploaded(t, importHost, body)
+ requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
+ got, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(importHost, begun.Seed.ID))
+ if err != nil || !bytes.Equal(got, body) {
+ t.Fatalf("snapshot = %d bytes, %v; want the upload", len(got), err)
+ }
+ if keys, _ := f.data.List(context.Background(), seeds.Prefix); len(keys) != 1 {
+ t.Fatalf("seeds/ = %v; want the one snapshot", keys)
+ }
+ f.requireClean(t)
+ events, err := f.fx.st.Events(OpsAccountID, fromApp, 50)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var rows []string
+ for _, e := range events {
+ if e.Action == "import-uploaded" {
+ rows = append(rows, e.Detail+" | "+e.Target+" | "+e.Address)
+ }
+ }
+ if want := "uploaded the snapshot for old.shop.example.com | old.shop.example.com | alice@keymail.dev"; len(rows) != 1 || rows[0] != want {
+ t.Fatalf("import-uploaded rows = %q, want [%q]", rows, want)
+ }
+}
+
+// The check runs after the answer, detached from the request (preflight
+// M6): the record waits in awaiting-upload, the spool sits in its own
+// from- directory (M5), and the reservation holds off a second upload.
+func TestImportUploadChecksAfterTheAnswer(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ type checking struct {
+ path string
+ deadline time.Duration
+ }
+ entered := make(chan checking, 1)
+ proceed := make(chan struct{})
+ old := checkSeedDBConsole
+ defer func() { checkSeedDBConsole = old }()
+ checkSeedDBConsole = func(ctx context.Context, path string) error {
+ d, _ := ctx.Deadline()
+ entered <- checking{path, time.Until(d)}
+ <-proceed
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
+ return old(ctx, path)
+ }
+ if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted || out["state"] != "checking" {
+ t.Fatalf("upload = %d %v", code, out)
+ }
+ c := <-entered
+ if dir := filepath.Dir(c.path); filepath.Base(c.path) != importHost+".db" || filepath.Dir(dir) != f.fx.srv.seedScratch.dir() ||
+ !strings.HasPrefix(filepath.Base(dir), seedScratchPrefix) {
+ t.Errorf("spool %s; want <scratch>/%s*/%s.db", c.path, seedScratchPrefix, importHost)
+ }
+ if c.deadline < 29*time.Minute || c.deadline > 30*time.Minute {
+ t.Errorf("the check's deadline is %v away, want 30 minutes", c.deadline)
+ }
+ rec, _ := f.record(t, importHost)
+ requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
+ if code, out := uploadImport(t, f, importHost, body); code != http.StatusTooManyRequests ||
+ str(t, out, "error") != "another copy or upload is running for shop. Try again when it finishes." {
+ t.Errorf("a second upload while the first is checked = %d %v, want 429", code, out)
+ }
+ close(proceed)
+ f.fx.srv.seedCopies.Wait()
+ rec, _ = f.record(t, importHost)
+ requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
+ f.requireClean(t)
+}
+
+// A length or hash that differs is refused before the answer, and the
+// import stays awaiting-upload, so the right file can still be sent.
+func TestImportUploadRefusesLengthAndHash(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ bad := append([]byte{}, body...)
+ bad[len(bad)-1] ^= 0xff
+ for name, b := range map[string][]byte{
+ "one byte long": append(append([]byte{}, body...), 'x'),
+ "one byte short": body[:len(body)-1],
+ "empty": {},
+ "wrong hash": bad,
+ } {
+ if code, out := uploadImport(t, f, importHost, b); code != http.StatusUnprocessableEntity || str(t, out, "error") != importMismatch {
+ t.Errorf("%s = %d %v, want 422", name, code, out)
+ }
+ f.fx.srv.seedCopies.Wait()
+ rec, _ := f.record(t, importHost)
+ requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+ }
+ requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
+}
+
+// A body far over its declared length is cut at bytes+1: the spool never
+// holds more than that (preflight M7).
+func TestImportUploadCutsALongBody(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ var spooled atomic.Int64
+ spooled.Store(-1)
+ defer func() { importSpoolHook = nil }()
+ importSpoolHook = func(path string) {
+ if st, err := os.Stat(path); err == nil {
+ spooled.Store(st.Size())
+ }
+ }
+ long := append(append([]byte{}, body...), bytes.Repeat([]byte{'x'}, 1<<20)...)
+ if code, out := uploadImport(t, f, importHost, long); code != http.StatusUnprocessableEntity || str(t, out, "error") != importMismatch {
+ t.Fatalf("a body 1 MiB long = %d %v, want 422", code, out)
+ }
+ if n := spooled.Load(); n < 0 || n > int64(len(body))+1 {
+ t.Fatalf("the spool held %d bytes; want at most %d", n, len(body)+1)
+ }
+ rec, _ := f.record(t, importHost)
+ requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+}
+
+func TestImportUploadOnlyOnce(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ f.begun(t, importHost, body)
+ f.uploaded(t, importHost, body)
+ if code, out := uploadImport(t, f, importHost, body); code != http.StatusConflict ||
+ str(t, out, "error") != "old.shop.example.com is uploaded. It takes an upload only while it waits for one." {
+ t.Fatalf("second upload = %d %v", code, out)
+ }
+ f.requireClean(t)
+}
+
+// A file of the right length and hash that is not a database is aborted,
+// saying why (SD-N52 as the seed's reason).
+func TestImportUploadRefusesANonDatabase(t *testing.T) {
+ f := newImportFixture(t)
+ junk := bytes.Repeat([]byte("j"), 100)
+ begun := f.begun(t, "junk.shop.example.com", junk)
+ requireImportState(t, f.uploaded(t, "junk.shop.example.com", junk), begun.Seed.ID, seeds.StateAborted, importNotADB)
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+}
+
+func TestImportUploadCheckDeadline(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ oldCheck, oldDeadline := checkSeedDBConsole, importCheckDeadline
+ defer func() { checkSeedDBConsole, importCheckDeadline = oldCheck, oldDeadline }()
+ importCheckDeadline = time.Nanosecond
+ checkSeedDBConsole = func(ctx context.Context, path string) error {
+ <-ctx.Done()
+ return ctx.Err()
+ }
+ requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importSlow)
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+}
+
+func TestImportUploadPutFailureAborts(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ f.fx.srv.cfg.SeedData = failingPuts{f.data}
+ requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+}
+
+// An object already at the seed's key is an earlier failed attempt for the
+// same seed: it is replaced.
+func TestImportUploadReplacesALeftover(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ key := seeds.SnapshotKey(importHost, begun.Seed.ID)
+ if _, err := f.data.Put(context.Background(), key, []byte("half of an earlier attempt"), ""); err != nil {
+ t.Fatal(err)
+ }
+ requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
+ if got, _, err := f.data.Get(context.Background(), key); err != nil || !bytes.Equal(got, body) {
+ t.Fatalf("snapshot = %q, %v; want the upload", got, err)
+ }
+}
+
+// The check panics: the seed is aborted with the retry sentence and the
+// console keeps serving.
+func TestImportUploadPanicAborts(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ old := checkSeedDBConsole
+ defer func() { checkSeedDBConsole = old }()
+ checkSeedDBConsole = func(context.Context, string) error { panic("a fault in the check") }
+ requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+ checkSeedDBConsole = old
+ other := f.begun(t, "new.shop.example.com", body)
+ requireImportState(t, f.uploaded(t, "new.shop.example.com", body), other.Seed.ID, seeds.StateUploaded, "")
+}
+
+// The uploaded write fails: the seed is aborted with the retry sentence.
+func TestImportUploadRecordWriteFailureAborts(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ begun := f.begun(t, importHost, body)
+ rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
+ f.fx.srv.cfg.Releases = rel
+ old := checkSeedDBConsole
+ defer func() { checkSeedDBConsole = old }()
+ checkSeedDBConsole = func(ctx context.Context, path string) error {
+ rel.armed.Store(true) // the next write of the record is uploaded
+ return old(ctx, path)
+ }
+ requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
+ f.requireClean(t)
+}
+
+func TestImportUploadRefusals(t *testing.T) {
+ f := newImportFixture(t)
+ body, _ := sqliteBytes(t)
+ f.begun(t, importHost, body)
+ for _, c := range []struct {
+ name, host string
+ code int
+ msg string
+ }{
+ {"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
+ {"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
+ {"an ordinary instance", fromSource, http.StatusConflict, "main.shop.example.com is not an import."},
+ } {
+ code, out := uploadImport(t, f, c.host, body)
+ if code != c.code || str(t, out, "error") != c.msg {
+ t.Errorf("%s = %d %v, want %d %q", c.name, code, out, c.code, c.msg)
+ }
+ }
+
+ // A --from seed is not an import either.
+ f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReserving)
+ if code, out := uploadImport(t, f, fromHost, body); code != http.StatusConflict || str(t, out, "error") != fromHost+" is not an import." {
+ t.Errorf("a --from seed = %d %v", code, out)
+ }
+
+ // A seed id that cannot name a key (preflight M8): nothing is put.
+ f.seedRecord(t, "bad.shop.example.com", "", "not-a-seed-id", seeds.KindImport, seeds.StateAwaitingUpload)
+ if code, out := uploadImport(t, f, "bad.shop.example.com", body); code != http.StatusBadGateway || str(t, out, "error") != importFailed {
+ t.Errorf("a bad seed id = %d %v", code, out)
+ }
+ f.requireNothingUploaded(t)
+
+ // A claim console refuses before it reads the record.
+ f.fx.srv.cfg.HostClaim = true
+ if code, out := uploadImport(t, f, importHost, body); code != http.StatusConflict ||
+ str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
+ t.Errorf("claims armed = %d %v", code, out)
+ }
+ f.fx.srv.cfg.HostClaim = false
+
+ // No data bucket (SD-N38).
+ data := f.fx.srv.cfg.SeedData
+ f.fx.srv.cfg.SeedData = nil
+ if code, out := uploadImport(t, f, importHost, body); code != http.StatusServiceUnavailable ||
+ str(t, out, "error") != "this console has no data bucket set up, so it cannot create a seeded instance." {
+ t.Errorf("no data bucket = %d %v", code, out)
+ }
+ f.fx.srv.cfg.SeedData = data
+
+ // No room in the scratch (SD-N27).
+ pool := f.fx.srv.seedScratch
+ f.fx.srv.seedScratch = newSeedScratch(pool.dir(), func(string) (int64, error) { return 1 << 30, nil })
+ if code, out := uploadImport(t, f, importHost, body); code != http.StatusInsufficientStorage ||
+ str(t, out, "error") != "the console has no room for this right now. Try again later." {
+ t.Errorf("no scratch = %d %v", code, out)
+ }
+ f.fx.srv.seedScratch = pool
+ f.fx.srv.seedCopies.Wait()
+ rec, _ := f.record(t, importHost)
+ if rec.Seed.State != seeds.StateAwaitingUpload {
+ t.Fatalf("a refused upload moved the import to %s", rec.Seed.State)
+ }
+ f.requireClean(t)
+ f.requireNothingUploaded(t)
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index 725d96432..3b9ae9827 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -689,6 +689,9 @@ func (s *Server) registerCLIRoutes() {
s.registerMovesPhase1Routes()
}
s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceDelete)
+ // The import door (seeded-instances spec §5): the instance role, like
+ // the create door that begins it.
+ s.cliRoute("PUT /api/cli/apps/{account}/{app}/instances/{host}/import", svcrole.Allow(svcrole.Instance), s.handleAPIImportUpload)
s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs", svcrole.Set{}, s.handleAPIRunCreate)
s.cliRoute("GET /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}", svcrole.Set{}, s.handleAPIRunGet)
s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/cancel", svcrole.Set{}, s.handleAPIRunCancel)
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index a0572ad3c..e99b02b58 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -1149,6 +1149,15 @@ var claimFixtures = map[string]claimFixture{
},
do: claimAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/web.go.tito.io/channel", `{"channel":"beta"}`),
}},
+ // Seeded instances wait for the claim's protocol (#250, seeded-instances
+ // spec §3.4): every import door refuses on a host-claim console before
+ // it reads the record (claimUnsupported).
+ "route PUT /api/cli/apps/{account}/{app}/instances/{host}/import": {refused: &claimRefusalFixture{
+ apps: claimRefs("bab/titogo"),
+ do: claimRefusedAPI(http.MethodPut, "/api/cli/apps/bab/titogo/instances/old.example.com/import", "SQLite format 3"),
+ status: http.StatusConflict,
+ msg: seedClaimConsoleRefusal().Msg,
+ }},
// carlos run waits for the claim's reservation protocol (#250): every
// run route refuses on a host-claim console (claimUnsupported).
"route POST /api/cli/apps/{account}/{app}/instances/{host}/runs": {refused: &claimRefusalFixture{
diff --git a/internal/console/claim_invariant_test.go b/internal/console/claim_invariant_test.go
index 996f11de3..6ca3f0786 100644
--- a/internal/console/claim_invariant_test.go
+++ b/internal/console/claim_invariant_test.go
@@ -106,6 +106,8 @@ var claimUnsupported = map[string]string{
"no operation gates both, so every transfer is refused",
claimMoveResolveUnavailableMsg: "the operator's door out of a stuck move (CARLOS_MOVES_PHASE1) has only its gate until " +
"the Moves resolution body lands; it answers this sentence and writes nothing",
+ seedClaimConsoleRefusal().Msg: "seeded instances need nested boxes, and the claim-armed path waits for #250 (seeded-instances spec §3.4); " +
+ "every import door refuses before it reads the record",
}
// claimLifecycleRoot is the classifier's NonApp reason for a lifecycle CAS
diff --git a/internal/console/server.go b/internal/console/server.go
index 5f3b37eb5..4650644c7 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -837,8 +837,9 @@ type Server struct {
claimWorkers map[string]func(context.Context)
dbHandleKey []byte
seedScratch *seedScratch
- // seedCopies counts --from copies running detached from their request
- // (apiinstanceseed.go), so a test can wait for one to finish.
+ // seedCopies counts --from copies and import uploads running detached
+ // from their request (apiinstanceseed.go, apiinstanceimport.go), so a
+ // test can wait for one to finish.
seedCopies sync.WaitGroup
cfg Config
store *Store
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go
index 6b536e572..8dadb72fe 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -472,6 +472,11 @@ var exemptFromLegacyPin = map[string]bool{
// nothing about the account; only what the deployment's database says
// about an address the caller chose.
"GET /api/cli/apps/{account}/{app}/geo": true,
+ // The import door (seeded-instances spec §6.1) is new access on the
+ // create door's footing: every import call takes the instance role, as
+ // the create that begins the import does. It did not exist before
+ // service principals.
+ "PUT /api/cli/apps/{account}/{app}/instances/{host}/import": true,
}
// TestInstanceRoleMatchesTodaysAppTokenReach is the compatibility pin of spec
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt
index ea5725bc6..a94eca4ac 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -27,6 +27,7 @@
../mail/aws.go|NewAWSProvisioner|iam.NewFromConfig(cfg)
../mail/aws.go|NewAWSProvisioner|sesv2.NewFromConfig(cfg)
../mail/aws.go|NewAWSProvisioner|sts.NewFromConfig(cfg)
+apiinstanceimport.go|putSnapshot|os.Open(path)
apiinstanceseed.go|uploadSnapshot|os.Open(path)
apiruns.go|handleAPIRunCreate|rec.ActiveRun.Open()
apiruns.go|writeRunObjects|os.Open(f.path)
diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt
index a42796eaa..b3bcac8ea 100644
--- a/internal/console/testdata/claim-routes.txt
+++ b/internal/console/testdata/claim-routes.txt
@@ -240,6 +240,7 @@ route PUT /api/cli/apps/{account}/{app}/config
route PUT /api/cli/apps/{account}/{app}/domains/{host}/paths
route PUT /api/cli/apps/{account}/{app}/email/events/push
route PUT /api/cli/apps/{account}/{app}/instances-manifest
+route PUT /api/cli/apps/{account}/{app}/instances/{host}/import
route PUT /api/cli/apps/{account}/{app}/pipeline
route PUT /api/cli/apps/{account}/{app}/sidecars-manifest
route PUT /api/cli/apps/{account}/{app}/telemetry