amadan

carlos / platform Public

host: Establish publishes outside the start's work, fenced per write and bounded (Task 6 fix round 1, finding 7, platform-1e P2)

StagePlatform and Commit made S3 requests inside work.With, which holds
the hostguard lock, so a stalled request delayed BeginFence. The data
bucket is now written through selectionStore: every write is fenced by
the start's captured selection immediately before the request (the
writerFence idiom, the lock taken for the check and released before the
request) and every request is bounded by hostbox.RequestLimit. The
publication runs outside With; With holds only the local effect, the
adoption record's removal.

Test: while the manifest's write stalls, another holder of the lock
(BeginFence's stand-in) acquires it and proceeds, and the establishment
completes once the write returns; the crash/retry test models a crash
between the commit and the local effect as the local effect's With
refusing.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 8ecd4515014ea270e9e82cd641049ae5126ca77e parent 342f84d
2 files changed, +158 −33
  • internal/host/movesfiles.go +56 −21
  • internal/host/movesfiles_test.go +102 −12
diff --git a/internal/host/movesfiles.go b/internal/host/movesfiles.go
index 4f524dbc7..f0eb656dd 100644
--- a/internal/host/movesfiles.go
+++ b/internal/host/movesfiles.go
@@ -456,7 +456,7 @@ func movesFilesOrigins(location, prefix string, m hostfiles.Manifest) ([]string,
// establishment of them (a retry after a crash past the commit: the same
// entries, the env bytes on disk, no adoption item, which the publisher
// then leaves as it is). Anything else refuses before any write.
-func (f *movesFiles) openPrepared(ctx context.Context, o hostclaim.Observed, gen, prefix string, env []byte) (*hostfiles.Publisher, error) {
+func (f *movesFiles) openPrepared(ctx context.Context, files release.Store, o hostclaim.Observed, gen, prefix string, env []byte) (*hostfiles.Publisher, error) {
if f.DataDir == "" {
return nil, hostwrite.ErrUnavailable
}
@@ -495,12 +495,12 @@ func (f *movesFiles) openPrepared(ctx context.Context, o hostclaim.Observed, gen
if prepared.Seq != 0 {
return nil, fmt.Errorf("%w: the prepared files manifest %s@%s is at sequence %d, not 0", hostbox.ErrChanged, key, pm.Version, prepared.Seq)
}
- cur, _, err := f.Files.Get(ctx, key)
+ cur, _, err := files.Get(ctx, key)
if err != nil {
return nil, err
}
if !bytes.Equal(cur, exact) {
- current, found, err := hostfiles.Load(ctx, f.Files, prefix)
+ current, found, err := hostfiles.Load(ctx, files, prefix)
if err != nil || !found {
return nil, fmt.Errorf("host: the files manifest %s: %v", key, err)
}
@@ -508,7 +508,7 @@ func (f *movesFiles) openPrepared(ctx context.Context, o hostclaim.Observed, gen
return nil, fmt.Errorf("%w: the files manifest %s is neither the prepared manifest %s nor this box's establishment of it", hostbox.ErrChanged, key, pm.Version)
}
}
- return hostfiles.Open(ctx, f.Files, prefix)
+ return hostfiles.Open(ctx, files, prefix)
}
// establishedOver reports whether cur is an establishment of prepared by
@@ -570,6 +570,37 @@ func (exactBytesStore) Delete(context.Context, string) error {
return errors.New("host: the prepared manifest's exact bytes are read only")
}
+// selectionStore is the data bucket as Establish writes it (fix round 1,
+// finding 7): every write is fenced by the start's captured selection
+// immediately before the request (work.With with no effect, the
+// writerFence idiom: the lock is taken for the check and released before
+// the request), and every request is bounded by hostbox.RequestLimit. No
+// request is made under the lock, so a stalled request never delays
+// BeginFence; a selection that no longer holds refuses the write.
+type selectionStore struct {
+ release.Store
+ fence func(context.Context) error
+}
+
+func (s selectionStore) bound(ctx context.Context) (context.Context, context.CancelFunc) {
+ return context.WithTimeout(ctx, hostbox.RequestLimit)
+}
+
+func (s selectionStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ ctx, cancel := s.bound(ctx)
+ defer cancel()
+ return s.Store.Get(ctx, key)
+}
+
+func (s selectionStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if err := s.fence(ctx); err != nil {
+ return "", err
+ }
+ ctx, cancel := s.bound(ctx)
+ defer cancel()
+ return s.Store.Put(ctx, key, body, ifMatch)
+}
+
// emptyManifest is the base of an empty start.
func emptyManifest() hostfiles.Manifest {
return hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}
@@ -590,11 +621,14 @@ func emptyManifest() hostfiles.Manifest {
// version; with none, the pin's never-published evidence or an error
// (hostfiles.ErrNoEvidence), never an empty bootstrap.
//
-// Then, under work: the env file's exact bytes on disk are staged as the
-// EnvItem, the adoption item is dropped, the revision is committed
-// (create-only for a new generation; compare-and-swap on P), and the local
-// adoption record is removed. Establish never writes the env file. An
-// error, a bucket that cannot be reached included, fails the wake.
+// Then the env file's exact bytes on disk are staged as the EnvItem, the
+// adoption item is dropped and the revision is committed (create-only for
+// a new generation; compare-and-swap on P), each bucket write fenced by
+// the start's selection immediately before it and bounded by the request
+// limit, never under the lock the work holds (selectionStore; fix round
+// 1, finding 7). Then, under work, the local adoption record is removed:
+// the one local effect. Establish never writes the env file. An error, a
+// bucket that cannot be reached included, fails the wake.
func (f *movesFiles) Establish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
if f == nil || f.Files == nil || f.Data == nil || f.Bucket == "" || f.Region == "" || f.Box == "" || f.FilesDir == nil || ctx == nil || work == nil {
return hostwrite.ErrUnavailable
@@ -613,15 +647,16 @@ func (f *movesFiles) Establish(ctx context.Context, o hostclaim.Observed, in Cla
return fmt.Errorf("host: the env file the files manifest must carry: %w", err)
}
prefix := movesFilesPrefix(h.Location, gen)
+ files := selectionStore{Store: f.Files, fence: writerFence(work, in.Identity)}
var p *hostfiles.Publisher
switch {
case in.Prepared:
- p, err = f.openPrepared(ctx, o, gen, prefix, env)
+ p, err = f.openPrepared(ctx, files, o, gen, prefix, env)
if err != nil {
return err
}
case in.Empty:
- p, err = hostfiles.NewFromCopy(f.Files, prefix, emptyManifest(), nil)
+ p, err = hostfiles.NewFromCopy(files, prefix, emptyManifest(), nil)
if err != nil {
return err
}
@@ -644,21 +679,21 @@ func (f *movesFiles) Establish(ctx context.Context, o hostclaim.Observed, in Cla
}
base, origins = emptyManifest(), nil
}
- p, err = hostfiles.NewFromCopy(f.Files, prefix, base, origins)
+ p, err = hostfiles.NewFromCopy(files, prefix, base, origins)
if err != nil {
return err
}
}
+ if err := p.StagePlatform(ctx, hostfiles.EnvItem, env); err != nil {
+ return err
+ }
+ if err := p.StagePlatformDelete(instancekey.AdoptionItem); err != nil {
+ return err
+ }
+ if _, err := p.Commit(ctx); err != nil {
+ return err
+ }
return work.With(ctx, in.Identity, func() error {
- if err := p.StagePlatform(ctx, hostfiles.EnvItem, env); err != nil {
- return err
- }
- if err := p.StagePlatformDelete(instancekey.AdoptionItem); err != nil {
- return err
- }
- if _, err := p.Commit(ctx); err != nil {
- return err
- }
adoption := filepath.Join(filepath.Dir(envPath), o.Host+instancekey.AdoptionSuffix)
if err := os.Remove(adoption); err != nil && !errors.Is(err, fs.ErrNotExist) {
return err
diff --git a/internal/host/movesfiles_test.go b/internal/host/movesfiles_test.go
index a38cfb097..c6181792a 100644
--- a/internal/host/movesfiles_test.go
+++ b/internal/host/movesfiles_test.go
@@ -13,6 +13,7 @@ import (
"strings"
"sync"
"testing"
+ "time"
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
@@ -98,11 +99,15 @@ type filesFixture struct {
files casBucketStore
dir string
f *movesFiles
- // work records whether the start's work ran and can refuse; workAfter
- // fails after the work's effect ran (a crash past it).
+ // work is the start's captured work, modelled as the lock it holds
+ // (the hostguard's): With takes mu, runs fn and releases it. workRan
+ // counts the calls; workErr refuses them, every call when workErrAt is
+ // 0, else the call with that 1-based index only (a crash at that
+ // point).
+ mu sync.Mutex
workRan int
workErr error
- workAfter error
+ workErrAt int
}
const (
@@ -111,14 +116,13 @@ const (
)
func (x *filesFixture) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ x.mu.Lock()
+ defer x.mu.Unlock()
x.workRan++
- if x.workErr != nil {
+ if x.workErr != nil && (x.workErrAt == 0 || x.workErrAt == x.workRan) {
return x.workErr
}
- if err := fn(); err != nil {
- return err
- }
- return x.workAfter
+ return fn()
}
func newFilesFixture(t *testing.T) *filesFixture {
@@ -449,7 +453,9 @@ func TestEstablishCopiesTheNewestValidManifestByExactVersion(t *testing.T) {
if err := x.f.Establish(ctx, o, launchInputs("g3"), x); err != nil {
t.Fatal(err)
}
- if x.workRan != 1 {
+ // One fence for the manifest's write (the unchanged env file takes no
+ // new object) and the local effect.
+ if x.workRan != 2 {
t.Fatalf("work ran %d times", x.workRan)
}
m, found := x.manifest("g3")
@@ -551,6 +557,88 @@ func TestEstablishKeepsOriginsPastAdmissionPruning(t *testing.T) {
}
}
+// stallingStore holds one key's write until released, signalling that it
+// is stalled: a slow S3 request.
+type stallingStore struct {
+ release.Store
+ key string
+ stalled chan struct{}
+ release chan struct{}
+ once sync.Once
+}
+
+func (s *stallingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if key == s.key {
+ s.once.Do(func() { close(s.stalled) })
+ select {
+ case <-s.release:
+ case <-ctx.Done():
+ return "", ctx.Err()
+ }
+ }
+ return s.Store.Put(ctx, key, body, ifMatch)
+}
+
+// Remote publication runs outside the start's work (fix round 1, finding
+// 7; platform-1e P2): each bucket write is fenced by a short With (the
+// selection still holds) and bounded by the request limit, and the lock
+// is released before the request is made; With holds only the local
+// effect. While the manifest's write stalls, another holder of the lock
+// (BeginFence's stand-in) acquires it and proceeds; the establishment
+// completes once the write returns.
+func TestEstablishPublishesOutsideTheStartsWork(t *testing.T) {
+ x := newFilesFixture(t)
+ ctx := context.Background()
+ x.writeEnv("K=1\n")
+ x.publish("g1", func(p *hostfiles.Publisher) {
+ if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
+ t.Fatal(err)
+ }
+ })
+ st := &stallingStore{Store: x.files, key: hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g2")), stalled: make(chan struct{}), release: make(chan struct{})}
+ x.f.Files = st
+ o := servingObserved("g1", "g1:uk:@root:uk")
+ done := make(chan error, 1)
+ go func() { done <- x.f.Establish(ctx, o, launchInputs("g2"), x) }()
+ select {
+ case <-st.stalled:
+ case err := <-done:
+ t.Fatalf("established before the manifest's write: %v", err)
+ case <-time.After(10 * time.Second):
+ t.Fatal("the manifest's write never began")
+ }
+ // The lock is free while the write stalls.
+ acquired := make(chan error, 1)
+ go func() {
+ acquired <- x.With(ctx, hostwrite.Identity{Config: "c", Release: "r"}, func() error { return nil })
+ }()
+ select {
+ case err := <-acquired:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the start's work is held across the stalled write: BeginFence would wait")
+ }
+ close(st.release)
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("the establishment never completed")
+ }
+ if m, found := x.manifest("g2"); !found || m.Seq != 2 { // g1's one entry, then the env object
+ t.Fatalf("g2: %+v %v", m, found)
+ }
+ // Two fences (the env object, the manifest), the probe above and the
+ // local effect.
+ if x.workRan != 4 {
+ t.Fatalf("work ran %d times", x.workRan)
+ }
+}
+
// The instance key rule (O1'): the manifest Establish leaves carries the
// env file with exactly the bytes on disk when it runs, staged and
// committed before it returns, and no adoption item; the local adoption
@@ -771,12 +859,14 @@ func TestEstablishOnThePreparedGenerationContinuesFromSequenceZero(t *testing.T)
o.Claim.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: "mv1", Phase: "probing", Exec: "uk-box"}
in := launchInputs("mv1")
in.Prepared = true
- // The first attempt crashes after the commit, before the local effect.
- x.workAfter = errors.New("crash after the commit")
+ // The first attempt crashes after the commit, before the local effect:
+ // the env object and the manifest are each fenced by one With, the
+ // local effect is the third, which refuses.
+ x.workErr, x.workErrAt = errors.New("crash after the commit"), 3
if err := x.f.Establish(ctx, o, in, x); err == nil {
t.Fatal("the crash was not reported")
}
- x.workAfter = nil
+ x.workErr, x.workErrAt = nil, 0
m, found := x.manifest("mv1")
if !found || m.Seq != 1 {
t.Fatalf("after the crashed attempt: %+v %v", m, found)