package console

// The leftover-replica rail on the two create doors (discussion #199). A
// deleted instance or sidecar leaves its replica at instances/<host> and any
// move staging at moves/<host>/, and both are keyed by HOST alone, so the next
// record created on that host would inherit the old database on its first
// wake or through `carlos db pull`. Until the full fix lands, both doors
// refuse a create on a host whose replica or staging prefix in this
// console's data bucket still holds an LTX file. These tests use real
// litestream replicas on disk behind the db door's replica seam, so "holds an
// LTX file" means exactly what the activator would restore from.

import (
	"context"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

type leftoverFx struct {
	*consoleFixture
	token string
	root  string        // the fake data bucket: one file replica per prefix
	opens *atomic.Int32 // how many times a door reached for a replica client
}

// leftoverFixture is a console in region ie with a data bucket configured,
// and titogo enabled for both instances and sidecars in ie. Neither manifest
// names a pool: a pooled record's data is on the pool box, which the rail
// skips (TestCreateSkipsTheCheckForPooledRecords).
func leftoverFixture(t *testing.T, code string) *leftoverFx {
	t.Helper()
	fx, token := instanceFixture(t, code)
	im := titoManifest()
	im.Pool = ""
	enableInstances(t, fx, "titogo", im)
	sm := titoSidecarManifest()
	sm.Pool = ""
	enableSidecars(t, fx, "titogo", sm)
	root := t.TempDir()
	fx.srv.cfg.HibernateBucket = "data-test"
	fx.srv.cfg.HibernatePrefix = "instances"
	fx.srv.cfg.Region = "ie"
	opens := new(atomic.Int32)
	fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
		opens.Add(1)
		return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
	}
	return &leftoverFx{consoleFixture: fx, token: token, root: root, opens: opens}
}

// leaveReplica writes a real litestream replica where a deleted record's
// activator left it.
func (fx *leftoverFx) leaveReplica(t *testing.T, host string) {
	t.Helper()
	dbdoortest.Replicate(t, filepath.Join(fx.root, activator.HibernatePrefix("instances", host)), 1)
}

// leaveStaging writes one where an abandoned move staged it.
func (fx *leftoverFx) leaveStaging(t *testing.T, host string) {
	t.Helper()
	dbdoortest.Replicate(t, filepath.Join(fx.root, activator.MovesPrefix(host)), 1)
}

// createDoor is one of the two doors, reduced to what the matrix needs.
type createDoor struct {
	name   string
	url    func(fx *consoleFixture) string
	body   string // a create the door accepts on a clean host, in region ie
	host   string // the host that create claims
	exists func(t *testing.T, fx *consoleFixture) bool
}

var leftoverDoors = []createDoor{
	{
		name: "instance",
		url:  func(fx *consoleFixture) string { return instancesURL(fx, "bab", "titogo") },
		body: `{"host":"acme.go.tito.io","region":"ie"}`,
		host: "acme.go.tito.io",
		exists: func(t *testing.T, fx *consoleFixture) bool {
			t.Helper()
			_, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "acme.go.tito.io")
			if err != nil && !errors.Is(err, os.ErrNotExist) {
				t.Fatal(err)
			}
			return err == nil
		},
	},
	{
		name: "sidecar",
		url:  func(fx *consoleFixture) string { return sidecarsURL(fx, "bab", "titogo") },
		body: `{"name":"bot","region":"ie"}`,
		host: botHost,
		exists: func(t *testing.T, fx *consoleFixture) bool {
			t.Helper()
			_, err := provision.GetSidecarRecord(context.Background(), fx.rel, "bab", "titogo", "bot")
			if err != nil && !errors.Is(err, os.ErrNotExist) {
				t.Fatal(err)
			}
			return err == nil
		},
	},
}

// A host whose replica prefix still holds LTX files is refused with 409,
// and the refusal writes no record.
func TestCreateRefusesLeftoverReplica(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-rep-"+d.name)
			fx.leaveReplica(t, d.host)
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") {
				t.Fatalf("create = %d %s, want the leftover-data 409", status, out)
			}
			if d.exists(t, fx.consoleFixture) {
				t.Fatal("a refused create wrote a record anyway")
			}
		})
	}
}

// Move staging counts too: a host whose only leftover is an abandoned
// moves/<host>/ is refused the same way, because a landing promotes that
// staging into the instance's prefix.
func TestCreateRefusesLeftoverMoveStaging(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-mov-"+d.name)
			fx.leaveStaging(t, d.host)
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") {
				t.Fatalf("create = %d %s, want the leftover-data 409", status, out)
			}
			if d.exists(t, fx.consoleFixture) {
				t.Fatal("a refused create wrote a record anyway")
			}
		})
	}
}

// A clean host is created as before, and the check really looked: both
// prefixes were opened.
func TestCreateAllowsACleanHost(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-clean-"+d.name)
			// Another host's leftovers are not this host's business.
			fx.leaveReplica(t, "other.go.tito.io")
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusCreated {
				t.Fatalf("create = %d %s, want 201", status, out)
			}
			if !d.exists(t, fx.consoleFixture) {
				t.Fatal("the create answered 201 and wrote no record")
			}
			if n := fx.opens.Load(); n != 2 {
				t.Fatalf("replica clients opened = %d, want 2 (the replica prefix and the staging prefix)", n)
			}
		})
	}
}

// A record homed in another region is not checked: its data lives in that
// region's bucket, which this console cannot read. The gap is documented at
// the rail; this pins that the rail does not refuse what it cannot see.
func TestCreateSkipsTheCheckInAnotherRegion(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-region-"+d.name)
			fx.srv.cfg.Region = "uk"
			fx.leaveReplica(t, d.host)
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusCreated {
				t.Fatalf("create = %d %s, want 201 (unchecked)", status, out)
			}
			if n := fx.opens.Load(); n != 0 {
				t.Fatalf("replica clients opened = %d for an out-of-region record, want 0", n)
			}
		})
	}
}

// A console with no data bucket creates as it always did, and never reaches
// for a replica client.
func TestCreateSkipsTheCheckWithNoDataBucket(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-nobucket-"+d.name)
			fx.srv.cfg.HibernateBucket = ""
			fx.leaveReplica(t, d.host)
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusCreated {
				t.Fatalf("create = %d %s, want 201 (unchecked)", status, out)
			}
			if n := fx.opens.Load(); n != 0 {
				t.Fatalf("replica clients opened = %d with no bucket, want 0", n)
			}
		})
	}
}

// failingReplica fails every listing, with a message carrying a bucket name
// the way a real S3 error does.
type failingReplica struct{}

func (failingReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
	return nil, errors.New("list secret-bucket: connection reset")
}

func (failingReplica) OpenLTXFile(context.Context, int, ltx.TXID, ltx.TXID, int64, int64) (io.ReadCloser, error) {
	return nil, errors.New("not used")
}

// erroringIterator fails on the first Next, the way the S3 client surfaces
// a paging error.
type erroringIterator struct{}

func (erroringIterator) Close() error        { return nil }
func (erroringIterator) Next() bool          { return false }
func (erroringIterator) Err() error          { return errors.New("page secret-bucket: timeout") }
func (erroringIterator) Item() *ltx.FileInfo { return nil }

type pagingFailReplica struct{ failingReplica }

func (pagingFailReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
	return erroringIterator{}, nil
}

// A check that cannot complete refuses with 503: never create on
// uncertainty. The store's own error, which names the bucket, stays in the
// log.
func TestCreateRefusesWhenTheCheckFails(t *testing.T) {
	seams := map[string]func(context.Context, string) (replicaReader, error){
		"client": func(context.Context, string) (replicaReader, error) {
			return nil, errors.New("resolve region for secret-bucket: no credentials")
		},
		"listing": func(context.Context, string) (replicaReader, error) { return failingReplica{}, nil },
		"paging":  func(context.Context, string) (replicaReader, error) { return pagingFailReplica{}, nil },
	}
	for _, d := range leftoverDoors {
		for seam, open := range seams {
			t.Run(d.name+"/"+seam, func(t *testing.T) {
				fx := leftoverFixture(t, "code-left-fail-"+d.name+"-"+seam)
				fx.srv.dbReplica = open
				status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
				if status != http.StatusServiceUnavailable || !strings.Contains(out, "could not check whether") {
					t.Fatalf("create = %d %s, want the 503 that refuses on uncertainty", status, out)
				}
				if strings.Contains(out, "secret-bucket") || strings.Contains(out, "data-test") {
					t.Fatalf("the refusal leaked a bucket name: %s", out)
				}
				if d.exists(t, fx.consoleFixture) {
					t.Fatal("a create that could not be checked wrote a record anyway")
				}
			})
		}
	}
}

// Every existing validation answers first: a create the door would refuse
// anyway never reaches for the bucket, and keeps its own answer.
func TestCreateValidatesBeforeTheCheck(t *testing.T) {
	// consoleRegion is the region the rail would check, so a check run
	// too early would reach for the bucket rather than skip.
	cases := []struct{ door, url, body, consoleRegion string }{
		{"instance", "instances", `{"host":"acme.elsewhere.io","region":"ie"}`, "ie"},
		{"instance", "instances", `{"host":"acme.go.tito.io","region":"mars"}`, "mars"},
		{"sidecar", "sidecars", `{"name":"../bot","region":"ie"}`, "ie"},
		{"sidecar", "sidecars", `{"name":"bot","region":"mars"}`, "mars"},
	}
	for _, c := range cases {
		t.Run(c.door+" "+c.body, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-order")
			fx.srv.cfg.Region = c.consoleRegion
			fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) {
				fx.opens.Add(1)
				return failingReplica{}, nil
			}
			u := fx.ts.URL + "/api/cli/apps/bab/titogo/" + c.url
			status, _, out := rawReq(t, http.MethodPost, u, fx.token, c.body)
			if status != http.StatusBadRequest {
				t.Fatalf("create = %d %s, want the door's own 400", status, out)
			}
			if n := fx.opens.Load(); n != 0 {
				t.Fatalf("an invalid create reached for the bucket %d time(s)", n)
			}
		})
	}
}

// Paths that create no record are untouched: deleting a record whose host
// holds a replica (the ordinary case) never consults the rail.
func TestDeleteNeverConsultsTheLeftoverCheck(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-del-"+d.name)
			if status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body); status != http.StatusCreated {
				t.Fatalf("create = %d %s", status, out)
			}
			fx.leaveReplica(t, d.host)
			fx.opens.Store(0)
			target := d.url(fx.consoleFixture) + "/" + d.host
			if d.name == "sidecar" {
				target = d.url(fx.consoleFixture) + "/bot"
			}
			if status, _, out := rawReq(t, http.MethodDelete, target, fx.token, ""); status >= 300 {
				t.Fatalf("delete = %d %s, want success", status, out)
			}
			if n := fx.opens.Load(); n != 0 {
				t.Fatalf("a delete reached for the bucket %d time(s)", n)
			}
		})
	}
}

// The residency rules the rail applies are resolveReplica's: in-region
// only, a pooled record's data is on the pool box, and a fleet placement is
// never this console's to read, whatever its own region string says.
func TestLeftoverCheckResidency(t *testing.T) {
	s := &Server{}
	s.cfg.HibernateBucket = "data-test"
	s.cfg.Region = "fleet/acme"
	for _, c := range []struct {
		region, pool, console string
		bucket                string
		want                  bool
	}{
		{"", "", "", "data-test", true},
		{"ie", "", "ie", "data-test", true},
		{"uk", "", "ie", "data-test", false},
		{"", "", "ie", "data-test", false},
		{"fleet/acme", "", "fleet/acme", "data-test", false},
		{"ie", "", "ie", "", false},
		{"ie", "tito", "ie", "data-test", false},
		{"", "tito", "", "data-test", false},
	} {
		s.cfg.Region, s.cfg.HibernateBucket = c.console, c.bucket
		if got := s.leftoverCheckApplies(c.region, c.pool); got != c.want {
			t.Errorf("record %q pool %q on console %q (bucket %q) = %v, want %v", c.region, c.pool, c.console, c.bucket, got, c.want)
		}
	}
}

// ---- review round 1 (Astra, 6cc86d34) -----------------------------------------

// fakeDataBucket is a ListObjectsV2 endpoint with real S3 prefix semantics,
// for the PRODUCTION replica client (lease.ReadReplicaClient through
// cachedReadReplica). The file replica the other tests use joins paths with
// filepath.Join, which forgives a doubled slash that S3 does not; this is
// where the exact keys the box writes, and the exact prefixes the console
// lists, meet.
type fakeDataBucket struct {
	mu       sync.Mutex
	keys     []string
	prefixes []string
}

func (b *fakeDataBucket) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	if r.Method != http.MethodGet || q.Get("list-type") != "2" {
		http.Error(w, "the leftover check only lists", http.StatusMethodNotAllowed)
		return
	}
	prefix := q.Get("prefix")
	b.mu.Lock()
	b.prefixes = append(b.prefixes, prefix)
	var keys []string
	for _, k := range b.keys {
		if strings.HasPrefix(k, prefix) {
			keys = append(keys, k)
		}
	}
	b.mu.Unlock()
	type obj struct {
		Key          string
		Size         int64
		LastModified string
	}
	type result struct {
		XMLName     xml.Name `xml:"ListBucketResult"`
		Name        string
		Prefix      string
		KeyCount    int
		IsTruncated bool
		Contents    []obj
	}
	res := result{Name: "data-test", Prefix: prefix, KeyCount: len(keys)}
	for _, k := range keys {
		res.Contents = append(res.Contents, obj{Key: k, Size: 100, LastModified: "2026-10-01T00:00:00Z"})
	}
	w.Header().Set("Content-Type", "application/xml")
	_ = xml.NewEncoder(w).Encode(res)
}

func (b *fakeDataBucket) listed() []string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return append([]string(nil), b.prefixes...)
}

// ltxKey is the key litestream's S3 client writes one LTX file under:
// <path>/<level as %04x>/<filename>.
func ltxKey(prefix string, level int) string {
	return fmt.Sprintf("%s/%04x/%s", strings.TrimSuffix(prefix, "/"), level, ltx.FormatFilename(1, 1))
}

// useFakeDataBucket points s at a fake S3 endpoint holding keys, through the
// production replica client.
func useFakeDataBucket(t *testing.T, s *Server, keys ...string) *fakeDataBucket {
	t.Helper()
	b := &fakeDataBucket{keys: keys}
	srv := httptest.NewServer(b)
	t.Cleanup(srv.Close)
	t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_SESSION_TOKEN", "")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
	s.cfg.HibernateBucket = "data-test"
	s.cfg.HibernateRegion = "us-east-1"
	s.cfg.HibernatePrefix = "instances"
	s.dbReplica = s.cachedReadReplica
	return b
}

// The production client is asked for exactly the prefixes the box writes:
// instances/<host>/<level>/ and moves/<host>/<level>/, levels 0 to 9, with
// one slash between each segment. A doubled slash lists nothing on S3.
func TestLeftoverCheckListsTheExactS3Prefixes(t *testing.T) {
	s := &Server{}
	b := useFakeDataBucket(t, s)
	held, err := s.hostHoldsReplicaData(context.Background(), "app.example.com")
	if err != nil || held {
		t.Fatalf("empty bucket = %v, %v; want false, nil", held, err)
	}
	var want []string
	for _, root := range []string{"instances/app.example.com", "moves/app.example.com"} {
		for level := 0; level <= 9; level++ {
			want = append(want, fmt.Sprintf("%s/%04x/", root, level))
		}
	}
	if got := b.listed(); strings.Join(got, "\n") != strings.Join(want, "\n") {
		t.Fatalf("listed prefixes:\n%s\nwant:\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
	}
}

// Data at any level counts, the snapshot level included, in either prefix;
// and a host is never answered with another host's data, even one its name
// is a string prefix of.
func TestLeftoverCheckThroughTheS3Client(t *testing.T) {
	const host = "app.example.co"
	for _, c := range []struct {
		name string
		keys []string
		want bool
	}{
		{"replica at L0", []string{ltxKey("instances/"+host, 0)}, true},
		{"snapshot only (L9)", []string{ltxKey("instances/"+host, 9)}, true},
		{"move staging at L0", []string{ltxKey(activator.MovesPrefix(host), 0)}, true},
		{"move staging snapshot only (L9)", []string{ltxKey(activator.MovesPrefix(host), 9)}, true},
		{"only a longer host's data", []string{
			ltxKey("instances/app.example.com", 0), ltxKey("instances/app.example.com", 9),
			ltxKey(activator.MovesPrefix("app.example.com"), 0),
		}, false},
		{"a lock file and nothing else", []string{"instances/" + host + "/lock.json"}, false},
	} {
		t.Run(c.name, func(t *testing.T) {
			s := &Server{}
			useFakeDataBucket(t, s, c.keys...)
			held, err := s.hostHoldsReplicaData(context.Background(), host)
			if err != nil || held != c.want {
				t.Fatalf("hostHoldsReplicaData = %v, %v; want %v, nil", held, err, c.want)
			}
		})
	}
}

// End to end through the production client: move staging the box wrote at
// moves/<host>/0000/ refuses both doors.
func TestCreateRefusesMoveStagingThroughTheS3Client(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-s3-"+d.name)
			useFakeDataBucket(t, fx.srv, ltxKey(activator.MovesPrefix(d.host), 0))
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") {
				t.Fatalf("create = %d %s, want the leftover-data 409", status, out)
			}
			if d.exists(t, fx.consoleFixture) {
				t.Fatal("a refused create wrote a record anyway")
			}
		})
	}
}

// writeDeadlineStore records the deadline of the context the record write
// ran under, and can fail reads of that record.
type writeDeadlineStore struct {
	release.Store
	key      string
	failGet  bool
	mu       sync.Mutex
	deadline time.Time
	puts     int
}

func (s *writeDeadlineStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if s.failGet && key == s.key {
		return nil, "", errors.New("get secret-bucket: connection reset")
	}
	return s.Store.Get(ctx, key)
}

func (s *writeDeadlineStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key {
		s.mu.Lock()
		s.deadline, _ = ctx.Deadline()
		s.puts++
		s.mu.Unlock()
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// recordKey is the bucket key a door's create writes.
func (d createDoor) recordKey() string {
	if d.name == "sidecar" {
		return provision.SidecarRecordKey("bab", "titogo", "bot")
	}
	return provision.RecordKey("bab", "titogo", d.host)
}

// slowReplica delays every listing, and remembers when the last one ended.
type slowReplica struct {
	replicaReader
	delay time.Duration
	last  *atomic.Int64
}

func (r slowReplica) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
	time.Sleep(r.delay)
	itr, err := r.replicaReader.LTXFiles(ctx, level, seek, useMetadata)
	r.last.Store(time.Now().UnixNano())
	return itr, err
}

// The record write gets a full budget of its own AFTER the check, rather
// than whatever the request's original deadline has left once validation and
// twenty listings have spent it.
func TestCreateWritesUnderAFreshDeadlineAfterTheCheck(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-deadline-"+d.name)
			last := new(atomic.Int64)
			fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
				return slowReplica{lsfile.NewReplicaClient(filepath.Join(fx.root, prefix)), 5 * time.Millisecond, last}, nil
			}
			st := &writeDeadlineStore{Store: fx.srv.cfg.Releases, key: d.recordKey()}
			fx.srv.cfg.Releases = st
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusCreated {
				t.Fatalf("create = %d %s, want 201", status, out)
			}
			checked := time.Unix(0, last.Load())
			if st.deadline.IsZero() {
				t.Fatal("the record write ran with no deadline")
			}
			if floor := checked.Add(instanceStoreTimeout); st.deadline.Before(floor) {
				t.Fatalf("record write deadline %s is %s short of a fresh budget after the check ended",
					st.deadline.Format(time.RFC3339Nano), floor.Sub(st.deadline))
			}
		})
	}
}

// A retried create whose first attempt landed (a lost response) gets the
// ordinary duplicate answer, unchanged, even though the live record's own
// replica now fills the prefix, and even when the bucket cannot be listed.
func TestCreateRetryOfAnExistingRecordIsTheOrdinaryDuplicate(t *testing.T) {
	dup := map[string]string{"instance": "instance acme.go.tito.io already exists", "sidecar": "sidecar bot already exists"}
	for _, d := range leftoverDoors {
		for _, seam := range []string{"replica", "failing listing"} {
			t.Run(d.name+"/"+seam, func(t *testing.T) {
				fx := leftoverFixture(t, "code-left-retry-"+d.name)
				if status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body); status != http.StatusCreated {
					t.Fatalf("first create = %d %s", status, out)
				}
				if seam == "replica" {
					fx.leaveReplica(t, d.host)
				} else {
					fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) { return failingReplica{}, nil }
				}
				before, _, err := fx.rel.Get(context.Background(), d.recordKey())
				if err != nil {
					t.Fatal(err)
				}
				status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
				if status != http.StatusConflict || !strings.Contains(out, dup[d.name]) {
					t.Fatalf("retry = %d %s, want the ordinary 409 %q", status, out, dup[d.name])
				}
				after, _, err := fx.rel.Get(context.Background(), d.recordKey())
				if err != nil || string(after) != string(before) {
					t.Fatalf("the retry changed the stored record (err %v)", err)
				}
			})
		}
	}
}

// A record read that fails is not "absent": the create is refused with the
// door's own failure, and nothing is written.
func TestCreateFailsClosedWhenTheRecordReadFails(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-getfail-"+d.name)
			st := &writeDeadlineStore{Store: fx.srv.cfg.Releases, key: d.recordKey(), failGet: true}
			fx.srv.cfg.Releases = st
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusBadGateway || strings.Contains(out, "secret-bucket") {
				t.Fatalf("create = %d %s, want the door's own 502 with no bucket name", status, out)
			}
			if st.puts != 0 {
				t.Fatalf("a create whose record read failed wrote the record %d time(s)", st.puts)
			}
		})
	}
}

// A pooled record's data lives on the pool's own box, not in this bucket
// (resolveReplica's rule), so the rail does not look, and leftovers here do
// not refuse it.
func TestCreateSkipsTheCheckForPooledRecords(t *testing.T) {
	for _, d := range leftoverDoors {
		t.Run(d.name, func(t *testing.T) {
			fx := leftoverFixture(t, "code-left-pool-"+d.name)
			enableInstances(t, fx.consoleFixture, "titogo", titoManifest())
			enableSidecars(t, fx.consoleFixture, "titogo", titoSidecarManifest())
			fx.leaveReplica(t, d.host)
			status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
			if status != http.StatusCreated {
				t.Fatalf("pooled create = %d %s, want 201 (unchecked)", status, out)
			}
			if n := fx.opens.Load(); n != 0 {
				t.Fatalf("replica clients opened = %d for a pooled record, want 0", n)
			}
		})
	}
}