amadan

carlos / platform Public

hostfiles: a commit whose response was lost is recognised even after later stages change the manifest

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 55af2876781496318d6928b7836a463913096dd1 parent 3a4ce1c
2 files changed, +83 −6
  • internal/hostfiles/hostfiles_test.go +56 −0
  • internal/hostfiles/publisher.go +27 −6
diff --git a/internal/hostfiles/hostfiles_test.go b/internal/hostfiles/hostfiles_test.go
index f83539e28..4e905c139 100644
--- a/internal/hostfiles/hostfiles_test.go
+++ b/internal/hostfiles/hostfiles_test.go
@@ -494,6 +494,7 @@ type lostResponseStore struct {
release.Store
armed bool
getFail bool // the next Get of the manifest fails once
+ putFail bool // the next Put of the manifest fails once, unapplied
}
func (l *lostResponseStore) Get(ctx context.Context, key string) ([]byte, string, error) {
@@ -505,6 +506,10 @@ func (l *lostResponseStore) Get(ctx context.Context, key string) ([]byte, string
}
func (l *lostResponseStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ if l.putFail && strings.HasSuffix(key, "manifest.json") {
+ l.putFail = false
+ return "", errors.New("connection refused")
+ }
got, err := l.Store.Put(ctx, key, body, etag)
if err == nil && l.armed && strings.HasSuffix(key, "manifest.json") {
l.armed = false
@@ -624,3 +629,54 @@ func TestThreeDeepOverlapKeepsTheNewestEntry(t *testing.T) {
t.Fatalf("a/b = %q, written %d, superseded %v", got, st.Written, st.Superseded)
}
}
+
+// A Stage between a commit whose response was lost and its retry changes
+// the body; the retry must still recognise the stored manifest as its own.
+func TestALostCommitResponseFollowedByAStageIsNotASupersede(t *testing.T) {
+ ctx := context.Background()
+ s := &lostResponseStore{Store: release.NewDirStore(t.TempDir())}
+ p := open(t, s)
+ stage(t, p, "x", "1")
+ s.armed = true
+ if _, err := p.Commit(ctx); err == nil || errors.Is(err, ErrSuperseded) {
+ t.Fatalf("first commit err = %v, want the transport error", err)
+ }
+ stage(t, p, "y", "2")
+ seq, err := p.Commit(ctx)
+ if err != nil || seq != 2 {
+ t.Fatalf("second commit = %d, %v; want 2, nil", seq, err)
+ }
+ m, _, err := Load(ctx, s, pfx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if m.Seq != 2 || m.Entries["x"].Seq != 1 || m.Entries["y"].Seq != 2 {
+ t.Fatalf("stored manifest = %+v, want x@1 and y@2 at seq 2", m)
+ }
+ stage(t, p, "z", "3")
+ if seq := commit(t, p); seq != 3 {
+ t.Fatalf("third commit seq = %d, want 3", seq)
+ }
+}
+
+// Two failed attempts against the same stale etag: only one can have been
+// applied, and whichever it was, the publisher recognises it.
+func TestTwoFailedCommitsThenAStageRecognisesTheAppliedOne(t *testing.T) {
+ ctx := context.Background()
+ s := &lostResponseStore{Store: release.NewDirStore(t.TempDir())}
+ p := open(t, s)
+ stage(t, p, "x", "1")
+ s.armed = true
+ if _, err := p.Commit(ctx); err == nil {
+ t.Fatal("first commit should fail")
+ }
+ stage(t, p, "y", "2")
+ s.putFail = true // the second attempt never reaches the store
+ if _, err := p.Commit(ctx); err == nil || errors.Is(err, ErrSuperseded) {
+ t.Fatalf("second commit err = %v, want the transport error", err)
+ }
+ stage(t, p, "z", "3")
+ if seq, err := p.Commit(ctx); err != nil || seq != 3 {
+ t.Fatalf("third commit = %d, %v; want 3, nil", seq, err)
+ }
+}
diff --git a/internal/hostfiles/publisher.go b/internal/hostfiles/publisher.go
index 36d9016f2..b1708f891 100644
--- a/internal/hostfiles/publisher.go
+++ b/internal/hostfiles/publisher.go
@@ -31,6 +31,11 @@ type Publisher struct {
etag string
next int64 // the last sequence number handed out; man.Seq is the last committed
dirty bool
+ // inflight holds, by sha256, every manifest body a failed Put may have
+ // applied since the last successful commit, with the sequence number it
+ // carried. Every one of those Puts named the same etag, so at most one
+ // of them can be in storage.
+ inflight map[string]int64
}
func Open(ctx context.Context, s release.Store, prefix string) (*Publisher, error) {
@@ -157,24 +162,40 @@ func (p *Publisher) Commit(ctx context.Context) (int64, error) {
}
etag, err := p.s.Put(ctx, ManifestKey(p.prefix), body, p.etag)
if errors.Is(err, release.ErrCASConflict) {
- // A lost response looks like this: the store applied our earlier Put,
- // the reply never arrived, and this retry's etag is now stale. If the
- // manifest holds exactly the bytes we were writing, it is ours.
+ // A lost response looks like this: the store applied an earlier Put,
+ // the reply never arrived, and this retry's etag is now stale. If
+ // the manifest holds exactly the bytes this Commit is writing, or
+ // the bytes an earlier failed Commit wrote, it is ours.
cur, curEtag, gerr := p.s.Get(ctx, ManifestKey(p.prefix))
if gerr != nil {
// Not knowing is not the same as being superseded: leave the
// staged state and etag alone so the next Commit repeats the check.
return 0, gerr
}
- if !bytes.Equal(cur, body) {
+ if bytes.Equal(cur, body) {
+ err, etag = nil, curEtag
+ } else if seq, ok := p.inflight[sum(cur)]; ok {
+ // An earlier attempt landed and later stages changed the body.
+ // Adopt it, then write the current body once on top of it.
+ p.man.Seq, p.etag, p.inflight = seq, curEtag, nil
+ etag, err = p.s.Put(ctx, ManifestKey(p.prefix), body, p.etag)
+ if errors.Is(err, release.ErrCASConflict) {
+ return 0, fmt.Errorf("%w: %s", ErrSuperseded, p.prefix)
+ }
+ } else {
return 0, fmt.Errorf("%w: %s", ErrSuperseded, p.prefix)
}
- err, etag = nil, curEtag
}
if err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ if p.inflight == nil {
+ p.inflight = map[string]int64{}
+ }
+ p.inflight[sum(body)] = out.Seq
+ }
return 0, err
}
- p.man.Seq, p.etag, p.dirty = out.Seq, etag, false
+ p.man.Seq, p.etag, p.dirty, p.inflight = out.Seq, etag, false, nil
return out.Seq, nil
}