amadan

carlos / platform Public

seeds: an aborted --from's row goes before its rollback, and the finish fences its replica delete (Plan 6 Task 4 fix round 1)

Box (review I1a): reconcile no longer rolls back an aborted --from seed
whose host still has a row on this box. It removes the row the way the
removal pass removes a retired record's row (Provisioned, this pass's
scope and kind, not a participant, through claimRetire), and a later
pass that finds no row rolls the seed back. An import's row is the
instance it replaces and is never removed here.

Console (review I1b, I2): finishSeed deletes no replica key while the
host's lease is held by lease.LockState's own rule (an expired lease
does not block); re-reads the record before listing the replica and
before every batch of deletes, stopping unless it still names this seed
rolled-back at the same etag; and deletes the record on the last etag
it verified. Also from the review: the data bucket is checked before
any teardown, a claim-armed console refuses, and tests for the prefix
boundary, the leftover-region skip, a truncated listing, the CAS retry,
and two assertions that could not fail.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b001b05410e21bdbe8eb2e756c6527f7393cd9d2 parent 677796b
5 files changed, +623 −21
  • internal/console/seedfinish.go +99 −21
  • internal/console/seedfinish_test.go +393 −0
  • internal/provision/reconcile.go +8 −0
  • internal/provision/seedreconcile.go +39 −0
  • internal/provision/seedreconcile_test.go +84 −0

CI passed — run details

diff --git a/internal/console/seedfinish.go b/internal/console/seedfinish.go
index 22b173641..016350cfe 100644
--- a/internal/console/seedfinish.go
+++ b/internal/console/seedfinish.go
@@ -12,8 +12,10 @@ import (
"fmt"
"log"
"os"
+ "time"
"github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
@@ -34,9 +36,13 @@ import (
// data bucket (M11). It is this seed's: the record is create-only and
// the create door's leftover-replica rail refused the host if anything
// was there, so only this seed's instance has written there since.
-// 4. The record, by CAS on the etag of the read that saw this seed in
-// rolled-back. It goes last because until then it holds the hostname,
-// so a re-import of the host never meets a leftover store.
+// That holds only while the record still names this seed, so the
+// record is re-read before the listing and before every batch of
+// deletes (review I2); and nothing is deleted while the host's lease is
+// held, because a row may still be writing (review I1).
+// 4. The record, by CAS on the etag of the last read that saw this seed
+// in rolled-back. It goes last because until then it holds the
+// hostname, so a re-import of the host never meets a leftover store.
//
// Every step is idempotent: a finish that fails part way answers an error
// with the record still rolled-back, and the next finish (the caller's
@@ -45,7 +51,14 @@ import (
// storeTeardownTimeout. Callers that finish inline on a request's behalf
// pass context.WithoutCancel of it, so a departing client cannot cut a
// teardown midway.
+//
+// A console with host claims armed refuses: its record deletes go through
+// the claim (retireClaimHostRecord), and the seed doors refuse such a
+// console anyway (seedRegionReady).
func (s *Server) finishSeed(ctx context.Context, sqid, app, host, seedID string) error {
+ if s.cfg.HostClaim {
+ return fmt.Errorf("finish seed %s of %s: this console has host claims armed, and finishes no seed", seedID, host)
+ }
if err := seeds.Check(host, seedID); err != nil {
return fmt.Errorf("finish seed %q of %q: %w", seedID, host, err)
}
@@ -69,19 +82,26 @@ func (s *Server) finishSeedOnce(ctx context.Context, sqid, app, host, seedID str
if rec.Seed == nil || rec.Seed.ID != seedID || rec.Seed.State != seeds.StateRolledBack {
return nil
}
- if err := s.finishSeedStore(ctx, sqid, app, host, seedID, rec.Environment); err != nil {
- return err
- }
+ // Checked before anything is destroyed: without it the store would be
+ // torn down and every later finish fail with the record holding the
+ // host.
if s.cfg.SeedData == nil {
return fmt.Errorf("finish seed %s of %s: this console has no data bucket to delete its snapshot from", seedID, host)
}
+ if err := s.finishSeedStore(ctx, sqid, app, host, seedID, rec.Environment); err != nil {
+ return err
+ }
if err := s.cfg.SeedData.Delete(ctx, seeds.SnapshotKey(host, seedID)); err != nil && !errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("delete the snapshot of seed %s of %s: %w", seedID, host, err)
}
if rec.Seed.Kind == seeds.KindReplica {
- if err := s.deleteSeedReplica(ctx, rec); err != nil {
+ var gone bool
+ if etag, gone, err = s.deleteSeedReplica(ctx, rec, etag); err != nil {
return err
}
+ if gone {
+ return nil // another finish deleted the record
+ }
}
if err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(sqid, app, host), etag); err != nil && !errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("delete the record of %s: %w", host, err)
@@ -146,30 +166,88 @@ func (s *Server) otherInstanceInEnvironment(ctx context.Context, sqid, app, host
return "", nil
}
+// seedReplicaBatch is how many replica keys are deleted between two
+// re-reads of the record.
+const seedReplicaBatch = 100
+
// deleteSeedReplica deletes every key under a --from seed's replica prefix
-// in the data bucket. The rail that makes this safe (createleftovers.go)
-// only checked a record this console can see; a record outside that
-// (another region, a pool, a fleet) was never checked, so its prefix here
-// is not known to be this seed's and is left alone.
-func (s *Server) deleteSeedReplica(ctx context.Context, rec *provision.Record) error {
+// in the data bucket, and answers the etag the record delete is to be
+// conditional on (the last read that saw this seed rolled-back), or gone
+// when the record was deleted meanwhile (another finish).
+//
+// The rail that makes the delete safe (createleftovers.go) only checked a
+// record this console can see; a record outside that (another region, a
+// pool, a fleet) was never checked, so its prefix here is not known to be
+// this seed's and is left alone.
+//
+// Nothing is deleted while the host's lease (lock.json under the prefix) is
+// held by the lease package's own rule (lease.LockState): a row this seed
+// started may still be running and writing there. A lease that is absent,
+// released or expired past its margin does not block, so a box that died
+// does not hold the host for ever. A held lease is an error, and the
+// record stays rolled-back for the next finish.
+func (s *Server) deleteSeedReplica(ctx context.Context, rec *provision.Record, etag string) (string, bool, error) {
+ sqid, app, host, seedID := rec.Account, rec.App, rec.Host, rec.Seed.ID
if !s.leftoverCheckApplies(rec.Region, rec.Pool) {
- log.Printf("console: finish seed %s of %s: its replica is outside this console's data bucket; not deleting it", rec.Seed.ID, rec.Host)
- return nil
+ log.Printf("console: finish seed %s of %s: its replica is outside this console's data bucket; not deleting it", seedID, host)
+ return etag, false, nil
+ }
+ // The read before the teardown can be a minute old: the host may have
+ // been finished and created again since, by a new instance that is
+ // writing here now.
+ etag, gone, err := s.seedRecordETag(ctx, sqid, app, host, seedID)
+ if gone || err != nil {
+ return "", gone, err
+ }
+ location := activator.HibernatePrefix(s.cfg.HibernatePrefix, host)
+ held, owner, expires, err := lease.LockState(ctx, s.cfg.HibernateBucket, location, s.cfg.HibernateRegion)
+ if err != nil {
+ return "", false, fmt.Errorf("read the lease of %s: %w", host, err)
+ }
+ if held {
+ return "", false, fmt.Errorf("finish seed %s of %s: its lease is held by %q until %s, so its replica is not deleted yet",
+ seedID, host, owner, expires.UTC().Format(time.RFC3339))
}
// The trailing slash keeps app.example.co from covering
// app.example.com's keys.
- prefix := activator.HibernatePrefix(s.cfg.HibernatePrefix, rec.Host) + "/"
+ prefix := location + "/"
keys, err := s.cfg.SeedData.List(ctx, prefix)
if err != nil {
- return fmt.Errorf("list the replica of %s: %w", rec.Host, err)
+ return "", false, fmt.Errorf("list the replica of %s: %w", host, err)
}
- for _, key := range keys {
- if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {
- return fmt.Errorf("delete the replica of %s: %w", rec.Host, err)
+ for start := 0; start < len(keys); start += seedReplicaBatch {
+ now, gone, err := s.seedRecordETag(ctx, sqid, app, host, seedID)
+ if gone || err != nil {
+ return "", gone, err
+ }
+ if now != etag {
+ return "", false, fmt.Errorf("finish seed %s of %s: the record changed while its replica was being deleted", seedID, host)
+ }
+ for _, key := range keys[start:min(start+seedReplicaBatch, len(keys))] {
+ if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return "", false, fmt.Errorf("delete the replica of %s: %w", host, err)
+ }
}
}
if len(keys) > 0 {
- log.Printf("console: finish seed %s of %s: deleted %d replica objects under %s", rec.Seed.ID, rec.Host, len(keys), prefix)
+ log.Printf("console: finish seed %s of %s: deleted %d replica objects under %s", seedID, host, len(keys), prefix)
}
- return nil
+ return etag, false, nil
+}
+
+// seedRecordETag re-reads host's record and answers its etag while it
+// still names seedID in rolled-back, or gone when there is no record. A
+// record that names anything else is an error, not "done": the finish only
+// says done when the seed's own record is gone.
+func (s *Server) seedRecordETag(ctx context.Context, sqid, app, host, seedID string) (etag string, gone bool, err error) {
+ rec, etag, err := provision.GetRecordETag(ctx, s.cfg.Releases, sqid, app, host)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ return "", true, nil
+ case err != nil:
+ return "", false, fmt.Errorf("read the record of %s: %w", host, err)
+ case rec.Seed == nil || rec.Seed.ID != seedID || rec.Seed.State != seeds.StateRolledBack:
+ return "", false, fmt.Errorf("finish seed %s of %s: the record no longer names this seed rolled-back", seedID, host)
+ }
+ return etag, false, nil
}
diff --git a/internal/console/seedfinish_test.go b/internal/console/seedfinish_test.go
index aae4a6342..b352416ac 100644
--- a/internal/console/seedfinish_test.go
+++ b/internal/console/seedfinish_test.go
@@ -3,13 +3,18 @@ package console
import (
"context"
"errors"
+ "fmt"
+ "net/http/httptest"
"os"
"path/filepath"
"slices"
+ "strings"
"testing"
"time"
"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
"github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
@@ -19,15 +24,57 @@ import (
// finishFixture is the --from fixture with private storage on and one data
// bucket, as production has it: seeds/ and every instance's replica in the
// same store, so the leftover-replica rail and the finish see one bucket.
+// The host's lease is read by the lease package itself (lease.LockState),
+// so it is served from leasetest's S3 server; leases answers for it.
func finishFixture(t *testing.T) fromFixture {
t.Helper()
f := newFromFixture(t, 3)
f.data = release.NewDirStore(f.root)
f.fx.srv.cfg.SeedData = f.data
enableInstances(t, f.fx, fromApp, fromManifest(true))
+ leases := leasetest.NewVersionedBucket()
+ leases.AddBucket(f.fx.srv.cfg.HibernateBucket)
+ handler, _ := leasetest.NewS3Server(leases)
+ srv := httptest.NewServer(handler)
+ t.Cleanup(srv.Close)
+ 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")
+ t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent")
+ t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent")
+ t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
+ f.fx.srv.cfg.HibernateRegion = "test-region"
+ finishLeases[f.fx.srv] = leases
+ t.Cleanup(func() { delete(finishLeases, f.fx.srv) })
return f
}
+// finishLeases is each finish fixture's lease bucket.
+var finishLeases = map[*Server]*leasetest.VersionedBucket{}
+
+// putLease writes host's lock.json as the litestream leaser does, expiring
+// at expires.
+func (f fromFixture) putLease(t *testing.T, host string, expires time.Time) {
+ t.Helper()
+ finishLeases[f.fx.srv].Put(f.fx.srv.cfg.HibernateBucket, "instances/"+host+"/lock.json",
+ []byte(`{"generation":1,"expires_at":"`+expires.UTC().Format(time.RFC3339Nano)+`","owner":"box-a"}`))
+}
+
+// putReplicaKeys writes n stand-in LTX objects under host's replica prefix.
+func (f fromFixture) putReplicaKeys(t *testing.T, host string, n int) []string {
+ t.Helper()
+ var keys []string
+ for i := 1; i <= n; i++ {
+ k := fmt.Sprintf("instances/%s/0000/%016x-%016x.ltx", host, i, i)
+ if _, err := f.data.Put(context.Background(), k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ keys = append(keys, k)
+ }
+ return keys
+}
+
// rollBack walks a seed the way a box does once it has been aborted: the
// abort, then the box's CAS to rolled-back.
func (f fromFixture) rollBack(t *testing.T, host, id string) {
@@ -344,6 +391,10 @@ type failingRecordList struct {
prefix string
}
+func (s failingRecordList) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ return release.DeleteIfMatch(ctx, s.Store, key, etag)
+}
+
func (s failingRecordList) List(ctx context.Context, prefix string) ([]string, error) {
if prefix == s.prefix {
return nil, errors.New("the bucket is unreachable")
@@ -421,6 +472,9 @@ func TestFinishSeedChecksTheSeedBeforeAnyKey(t *testing.T) {
if _, err := f.data.Put(ctx, sentinel, []byte("someone else's"), ""); err != nil {
t.Fatal(err)
}
+ // A rolled-back record carrying the bad id: without the check, the
+ // finish would match it and delete the sentinel.
+ f.seedRecord(t, fromHost, "canary", "sd_bad", seeds.KindImport, seeds.StateRolledBack)
for _, c := range []struct{ host, id string }{{fromHost, "sd_bad"}, {"../shop", newSeedID(t)}} {
if err := f.fx.srv.finishSeed(ctx, "bab", fromApp, c.host, c.id); !errors.Is(err, seeds.ErrBadID) && !errors.Is(err, seeds.ErrBadHost) {
t.Fatalf("finishSeed(%q, %q) = %v; want a seeds.Check error", c.host, c.id, err)
@@ -438,3 +492,342 @@ func TestFinishSeedWithNoRecord(t *testing.T) {
t.Fatal(err)
}
}
+
+// rolledBackFrom writes a --from record rolled back on host, with replica
+// keys under its prefix, and answers its seed id and those keys.
+func (f fromFixture) rolledBackFrom(t *testing.T, host string) (string, []string) {
+ t.Helper()
+ id := newSeedID(t)
+ f.seedRecord(t, host, "canary", id, seeds.KindReplica, seeds.StateRolledBack)
+ return id, f.putReplicaKeys(t, host, 3)
+}
+
+// Review I1 (b): a lease that is present and unexpired by the lease
+// package's own rule blocks the replica step. Nothing is deleted, the
+// error is retryable and the record stays rolled-back. An absent lease, or
+// one expired past its margin (a box that died), does not block.
+func TestFinishSeedWaitsForTheHostsLease(t *testing.T) {
+ // Expiries are relative to the moment each case writes its lease: the
+ // margin is seconds, and earlier cases take most of one.
+ for name, c := range map[string]struct {
+ expiresIn *time.Duration
+ blocked bool
+ }{
+ "held": {expiresIn: ptr(time.Minute), blocked: true},
+ "just expired": {expiresIn: ptr(-lease.RenewMargin / 2), blocked: true},
+ "expired past margin": {expiresIn: ptr(-2 * lease.RenewMargin)},
+ "absent": {},
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := finishFixture(t)
+ id, keys := f.rolledBackFrom(t, fromHost)
+ if c.expiresIn != nil {
+ f.putLease(t, fromHost, time.Now().Add(*c.expiresIn))
+ }
+ err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id)
+ if !c.blocked {
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := replicaKeys(t, f, fromHost); len(got) != 0 {
+ t.Fatalf("replica after the finish: %v", got)
+ }
+ f.requireRecordGone(t, fromHost)
+ return
+ }
+ if err == nil {
+ t.Fatal("a held lease answered nil")
+ }
+ if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
+ t.Fatalf("replica under a held lease:\n before %v\n after %v", keys, got)
+ }
+ if got, err := f.record(t, fromHost); err != nil || got.Seed.State != seeds.StateRolledBack {
+ t.Fatalf("record under a held lease = %+v, %v; want it kept, rolled-back", got, err)
+ }
+ })
+ }
+}
+
+func ptr[T any](v T) *T { return &v }
+
+// recreatingData runs recreate once: before the first Delete whose key
+// onDelete matches, or after the first List whose prefix onList matches.
+type recreatingData struct {
+ release.Store
+ onDelete, onList func(key string) bool
+ recreate func()
+ done bool
+}
+
+func (d *recreatingData) fire(match func(string) bool, key string) {
+ if !d.done && match != nil && match(key) {
+ d.done = true
+ d.recreate()
+ }
+}
+
+func (d *recreatingData) Delete(ctx context.Context, key string) error {
+ d.fire(d.onDelete, key)
+ return d.Store.Delete(ctx, key)
+}
+
+func (d *recreatingData) List(ctx context.Context, prefix string) ([]string, error) {
+ keys, err := d.Store.List(ctx, prefix)
+ d.fire(d.onList, prefix)
+ return keys, err
+}
+
+// recreateHost deletes host's record and writes a new, seedless instance
+// of it, as another finish and then a new create would.
+func (f fromFixture) recreateHost(t *testing.T, host string) func() {
+ return func() {
+ ctx := context.Background()
+ if err := provision.DeleteRecord(ctx, f.fx.rel, "bab", fromApp, host); err != nil {
+ t.Error(err)
+ }
+ if err := provision.CreateRecord(ctx, f.fx.rel, provision.Record{Schema: 1, Host: host, Slug: "canary",
+ App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil {
+ t.Error(err)
+ }
+ }
+}
+
+// Review I2: the replica delete is fenced on the record. A record deleted
+// and created again for the same host, between the store teardown and the
+// replica step, or while the replica is listed, keeps every replica key:
+// they may be the new instance's. The finish answers an error (the
+// record it was finishing is not gone in the sense that matters: the host
+// has a record), and the new record is untouched.
+func TestFinishSeedNeverDeletesANewInstancesReplica(t *testing.T) {
+ for name, wrap := range map[string]func(f fromFixture) *recreatingData{
+ "before the replica step": func(f fromFixture) *recreatingData {
+ return &recreatingData{Store: f.data, recreate: f.recreateHost(t, fromHost),
+ onDelete: func(key string) bool { return strings.HasPrefix(key, seeds.Prefix) }}
+ },
+ "while it is listed": func(f fromFixture) *recreatingData {
+ return &recreatingData{Store: f.data, recreate: f.recreateHost(t, fromHost),
+ onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") }}
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := finishFixture(t)
+ id, keys := f.rolledBackFrom(t, fromHost)
+ d := wrap(f)
+ f.fx.srv.cfg.SeedData = d
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
+ t.Fatal("a re-created host answered nil")
+ }
+ if !d.done {
+ t.Fatal("the record was never re-created: the test did not reach its window")
+ }
+ if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
+ t.Fatalf("the new instance's replica:\n before %v\n after %v", keys, got)
+ }
+ if got, err := f.record(t, fromHost); err != nil || got.Seed != nil {
+ t.Fatalf("the new record = %+v, %v; want it untouched", got, err)
+ }
+ // The sweep's next finish of the old seed leaves the new
+ // instance alone and is done.
+ f.fx.srv.cfg.SeedData = f.data
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
+ t.Fatal(err)
+ }
+ if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
+ t.Fatalf("the retry touched the new instance's replica: %v", got)
+ }
+ })
+ }
+}
+
+// A record rewritten in place while the replica is listed (an ordinary
+// writer, the same seed still rolled-back) stops the deletes too: the
+// finish only deletes under the record it read. The next finish resumes.
+func TestFinishSeedStopsWhenTheRecordChangesMidReplica(t *testing.T) {
+ f := finishFixture(t)
+ id, keys := f.rolledBackFrom(t, fromHost)
+ key := provision.RecordKey("bab", fromApp, fromHost)
+ f.fx.srv.cfg.SeedData = &recreatingData{Store: f.data,
+ onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") },
+ recreate: func() {
+ ctx := context.Background()
+ raw, etag, err := f.fx.rel.Get(ctx, key)
+ if err == nil {
+ _, err = f.fx.rel.Put(ctx, key, append(raw, '\n'), etag)
+ }
+ if err != nil {
+ t.Error(err)
+ }
+ }}
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
+ t.Fatal("a record changed mid-replica answered nil")
+ }
+ if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
+ t.Fatalf("replica after the record changed:\n before %v\n after %v", keys, got)
+ }
+ f.fx.srv.cfg.SeedData = f.data
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
+ t.Fatal(err)
+ }
+ if got := replicaKeys(t, f, fromHost); len(got) != 0 {
+ t.Fatalf("replica after the resumed finish: %v", got)
+ }
+ f.requireRecordGone(t, fromHost)
+}
+
+// A record deleted (not re-created) during the replica step was finished
+// by another finish: nil, and nothing more is deleted.
+func TestFinishSeedStopsWhenAnotherFinishDeletedTheRecord(t *testing.T) {
+ f := finishFixture(t)
+ id, keys := f.rolledBackFrom(t, fromHost)
+ f.fx.srv.cfg.SeedData = &recreatingData{Store: f.data,
+ onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") },
+ recreate: func() {
+ if err := provision.DeleteRecord(context.Background(), f.fx.rel, "bab", fromApp, fromHost); err != nil {
+ t.Error(err)
+ }
+ }}
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
+ t.Fatal(err)
+ }
+ if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
+ t.Fatalf("replica after the record went: %v", got)
+ }
+}
+
+// The prefix stops at the host: a host that extends this one's name keeps
+// its replica.
+func TestFinishSeedReplicaPrefixStopsAtTheHost(t *testing.T) {
+ f := finishFixture(t)
+ id, _ := f.rolledBackFrom(t, fromHost)
+ sibling := f.putReplicaKeys(t, fromHost+".au", 2)
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
+ t.Fatal(err)
+ }
+ if got := replicaKeys(t, f, fromHost); len(got) != 0 {
+ t.Fatalf("replica after the finish: %v", got)
+ }
+ if got := replicaKeys(t, f, fromHost+".au"); !slices.Equal(got, sibling) {
+ t.Fatalf("the sibling host's replica:\n before %v\n after %v", sibling, got)
+ }
+}
+
+// A replica outside this console's data bucket (another region) is not
+// known to be the seed's, and is left alone; the record still goes.
+func TestFinishSeedLeavesAReplicaOutsideItsBucket(t *testing.T) {
+ f := finishFixture(t)
+ id := newSeedID(t)
+ rec := provision.Record{Schema: 1, Host: fromHost, Slug: "canary", App: fromApp, Account: "bab", Channel: "stable",
+ Environment: "canary", Region: "elsewhere", CreatedAt: time.Now().UTC(),
+ Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateRolledBack, By: "bob@keymail.dev", At: time.Now().UTC()}}
+ if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {
+ t.Fatal(err)
+ }
+ keys := f.putReplicaKeys(t, fromHost, 2)
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
+ t.Fatal(err)
+ }
+ if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
+ t.Fatalf("a replica outside this bucket was touched: %v", got)
+ }
+ f.requireRecordGone(t, fromHost)
+}
+
+// More instances than one listing holds: the finish cannot tell whether
+// the store is shared, so it does nothing and answers an error.
+func TestFinishSeedRefusesATruncatedListing(t *testing.T) {
+ f := finishFixture(t)
+ id := newSeedID(t)
+ f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
+ f.markedStore(t, id)
+ for i := range instanceListMax {
+ if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: fmt.Sprintf("i%d.shop.example.com", i),
+ Slug: fmt.Sprintf("i%d", i), App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
+ t.Fatal("a truncated listing answered nil")
+ }
+ if f.canaryStore(t) == nil {
+ t.Fatal("the store went on a truncated listing")
+ }
+ if _, err := f.record(t, fromHost); err != nil {
+ t.Fatalf("the record went on a truncated listing: %v", err)
+ }
+}
+
+// touchOnDelete rewrites the record once (an ordinary writer's update)
+// just before the finish's conditional delete, so that delete loses.
+type touchOnDelete struct {
+ release.Store
+ key string
+ touched bool
+}
+
+func (s *touchOnDelete) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ if key == s.key && !s.touched {
+ s.touched = true
+ raw, cur, err := s.Store.Get(ctx, key)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Store.Put(ctx, key, append(raw, '\n'), cur); err != nil {
+ return err
+ }
+ }
+ return release.DeleteIfMatch(ctx, s.Store, key, etag)
+}
+
+// A record changed between the finish's read and its delete is read again
+// once, and the finish completes.
+func TestFinishSeedRetriesALostRecordDelete(t *testing.T) {
+ f := finishFixture(t)
+ id := newSeedID(t)
+ f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
+ rel := &touchOnDelete{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, fromHost)}
+ f.fx.srv.cfg.Releases = rel
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
+ t.Fatal(err)
+ }
+ if !rel.touched {
+ t.Fatal("the record was never touched: the test did not reach its window")
+ }
+ f.requireRecordGone(t, fromHost)
+}
+
+// A console with no data bucket refuses before anything is destroyed: the
+// store this seed made still stands, with the record.
+func TestFinishSeedNeedsADataBucketBeforeTheTeardown(t *testing.T) {
+ f := finishFixture(t)
+ id := newSeedID(t)
+ f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
+ f.markedStore(t, id)
+ f.fx.srv.cfg.SeedData = nil
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
+ t.Fatal("no data bucket answered nil")
+ }
+ if f.canaryStore(t) == nil {
+ t.Fatal("the store was torn down by a finish that could not complete")
+ }
+ if _, err := f.record(t, fromHost); err != nil {
+ t.Fatalf("the record went: %v", err)
+ }
+}
+
+// A console with host claims armed finishes no seed.
+func TestFinishSeedRefusesAClaimConsole(t *testing.T) {
+ f := finishFixture(t)
+ id := newSeedID(t)
+ f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
+ f.markedStore(t, id)
+ f.fx.srv.cfg.HostClaim = true
+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
+ t.Fatal("a claim console answered nil")
+ }
+ if f.canaryStore(t) == nil {
+ t.Fatal("a claim console tore the store down")
+ }
+ if _, err := f.record(t, fromHost); err != nil {
+ t.Fatalf("a claim console deleted the record: %v", err)
+ }
+}
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go
index 87a9fee57..d8c5428a4 100644
--- a/internal/provision/reconcile.go
+++ b/internal/provision/reconcile.go
@@ -27,6 +27,7 @@ import (
"github.com/carlosframework/platform/internal/domains"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/seeds"
)
// ReconcileConfig is everything one pass reads its world through — the same
@@ -1102,6 +1103,13 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
case seedSkip:
continue
case seedFinishAbort:
+ // A --from seed this box minted is still running: its row
+ // goes first, and a later pass that finds no row rolls the
+ // seed back (seedreconcile.go).
+ if ok && rec.Seed.Kind == seeds.KindReplica {
+ retireAbortedSeedRow(ctx, reg, cfg, rec, existing)
+ continue
+ }
finishAbortedSeed(ctx, s, cfg, rec)
continue
}
diff --git a/internal/provision/seedreconcile.go b/internal/provision/seedreconcile.go
index 4475952c9..ad41d2f2e 100644
--- a/internal/provision/seedreconcile.go
+++ b/internal/provision/seedreconcile.go
@@ -6,6 +6,7 @@ import (
"log"
"os"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/seeds"
)
@@ -73,6 +74,44 @@ func seedAction(rec Record, hasRow bool, cfg ReconcileConfig, nested func() (boo
return seedSkip
}
+// retireAbortedSeedRow removes the row of an aborted --from seed that this
+// box minted (task-4 review I1). The seed can be aborted after its publish
+// landed (a lost publish answer), and then a box may have minted the row and
+// be running it, holding the host's lease and writing its replica. The
+// console's finish deletes that replica, so the row must stop first. The
+// record stays aborted this pass; a later pass that finds no row rolls the
+// seed back (finishAbortedSeed).
+//
+// It removes the row exactly as the removal pass removes a row whose record
+// is gone, with the same authority: a Provisioned row of this pass's own
+// kind and scope, not a running operation's participant, through
+// claimRetire (reg.Delete, which the activator follows by stopping it). It
+// also requires the row to be this record's (account and app). A row it
+// may not remove is left alone, logged, and the seed stays aborted.
+//
+// Only --from seeds: an import's row is the instance the import replaces,
+// which its rollback restores (Plan 7) and never removes.
+func retireAbortedSeedRow(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record, rt registry.Route) {
+ if rt.Kind == SidecarKind || !rt.Provisioned || !cfg.inScope(rt.Region, rt.Pool, everywhereRoute(rt)) ||
+ rt.Account != rec.Account || rt.App != rec.App {
+ log.Printf("provision: %s: aborted seed %s has a row this pass may not remove; leaving the seed aborted", rec.Host, rec.Seed.ID)
+ return
+ }
+ if cfg.KeepParticipant != nil && cfg.KeepParticipant(rt.Host) {
+ log.Printf("provision: %s: aborted seed %s: not removing its row: participant of a running operation", rec.Host, rec.Seed.ID)
+ return
+ }
+ if cfg.DryRun {
+ log.Printf("provision: dry-run: would remove %s route %s (its seed %s was aborted)", rt.Kind, rt.Host, rec.Seed.ID)
+ return
+ }
+ if err := cfg.claimRetire(ctx, reg, rt); err != nil {
+ log.Printf("provision: remove %s route %s (its seed %s was aborted): %v", rt.Kind, rt.Host, rec.Seed.ID, err)
+ return
+ }
+ log.Printf("provision: removed %s route %s (its seed %s was aborted); the next pass rolls it back", rt.Kind, rt.Host, rec.Seed.ID)
+}
+
// finishAbortedSeed is §3.2's no-journal path for an aborted seed: delete
// this seed's snapshot (only this seed's id, codex R4-1), then CAS the
// record to rolled-back. Idempotent: an absent snapshot and a lost CAS are
diff --git a/internal/provision/seedreconcile_test.go b/internal/provision/seedreconcile_test.go
index 080dae8aa..56a2fafcf 100644
--- a/internal/provision/seedreconcile_test.go
+++ b/internal/provision/seedreconcile_test.go
@@ -157,3 +157,87 @@ func TestReconcileDoesNotMintASeedThatWouldNotHibernate(t *testing.T) {
t.Errorf("minted %+v", rt)
}
}
+
+// A --from seed aborted after this box minted its row (a lost publish
+// answer): the row is removed first, the way a retired record's row is, and
+// the seed stays aborted with its snapshot. The next pass, which finds no
+// row, rolls it back.
+func TestReconcileRemovesAnAbortedFromSeedsRowBeforeRollingBack(t *testing.T) {
+ s, sqid := setup(t)
+ data := dirStore(t)
+ rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateReady)
+ snap := seeds.SnapshotKey(rec.Host, rec.Seed.ID)
+ if _, err := data.Put(ctx, snap, []byte("db"), ""); err != nil {
+ t.Fatal(err)
+ }
+ reg := regWith(t)
+ if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
+ t.Fatal(err)
+ }
+ mustGet(t, reg, rec.Host)
+ if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
+ t.Fatal(err)
+ }
+
+ if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
+ t.Fatal(err)
+ }
+ if rt, _ := reg.Get(rec.Host); rt != nil {
+ t.Fatalf("the aborted seed's row is still up: %+v", rt)
+ }
+ got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
+ if err != nil || got.Seed.State != seeds.StateAborted {
+ t.Fatalf("record after the row's removal = %+v, %v; want still aborted", got.Seed, err)
+ }
+ if _, _, err := data.Get(ctx, snap); err != nil {
+ t.Fatalf("the snapshot went with the row: %v", err)
+ }
+
+ if err := Reconcile(ctx, s, reg, nestedSeedCfg(data)); err != nil {
+ t.Fatal(err)
+ }
+ got, err = GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
+ if err != nil || got.Seed.State != seeds.StateRolledBack {
+ t.Fatalf("record on the pass with no row = %+v, %v; want rolled-back", got.Seed, err)
+ }
+ if _, _, err := data.Get(ctx, snap); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("snapshot after the rollback: %v, want it deleted", err)
+ }
+ if rt, _ := reg.Get(rec.Host); rt != nil {
+ t.Fatalf("a rolled-back seed was minted again: %+v", rt)
+ }
+}
+
+// A row this pass has no authority over (not Provisioned: hand-added) is
+// left alone, and the seed stays aborted while it stands.
+func TestReconcileLeavesAnAbortedSeedsUnownedRow(t *testing.T) {
+ s, sqid := setup(t)
+ rec := seededRecord(t, s, sqid, seeds.KindReplica, seeds.StateAborted)
+ reg := regWith(t, registry.Route{Host: rec.Host, Socket: "/run/carlos/hand.sock", DBPath: "/data/hand.db",
+ App: rec.App, Account: rec.Account, Region: "ie", Pool: "tito"})
+ if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
+ t.Fatal(err)
+ }
+ mustGet(t, reg, rec.Host)
+ got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
+ if err != nil || got.Seed.State != seeds.StateAborted {
+ t.Fatalf("record = %+v, %v; want still aborted while the row stands", got.Seed, err)
+ }
+}
+
+// An import's row is the instance it replaces: an aborted import never
+// removes it, and rolls back as before (its swap's rollback is Plan 7's).
+func TestReconcileNeverRemovesAnAbortedImportsRow(t *testing.T) {
+ s, sqid := setup(t)
+ rec := seededRecord(t, s, sqid, seeds.KindImport, seeds.StateAborted)
+ reg := regWith(t, registry.Route{Host: rec.Host, Socket: DefaultSocket(rec.Host), DBPath: DefaultDBPath("/data", rec.Host),
+ App: rec.App, Account: rec.Account, Region: "ie", Pool: "tito", Provisioned: true})
+ if err := Reconcile(ctx, s, reg, nestedSeedCfg(dirStore(t))); err != nil {
+ t.Fatal(err)
+ }
+ mustGet(t, reg, rec.Host)
+ got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
+ if err != nil || got.Seed.State != seeds.StateRolledBack {
+ t.Fatalf("record = %+v, %v; want rolled-back", got.Seed, err)
+ }
+}