amadan

carlos / platform Public

moves: a table write the server rejects (a non-CAS 4xx) is an answer: it failed, is never resent, and its owner removes its registration

Ruling on Task 2a's concern 5. S3Store.PutOnce wraps a 4xx other than a
lost condition, 408, 429 and a missing bucket's 404 as
release.ErrPutRejected; PutTable and Resend count it as answered. Only a
timeout, a transport error or a 5xx stays uncertain and is resent.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 45e0b220c6430017edef61e9ffc2b5ecc1348856 parent 2aefe5c
4 files changed, +130 −3
  • internal/moves/tablewrite.go +8 −3
  • internal/moves/tablewrite_test.go +57 −0
  • internal/release/putonce_test.go +36 −0
  • internal/release/store.go +29 −0
diff --git a/internal/moves/tablewrite.go b/internal/moves/tablewrite.go
index de7d33415..3e1249d03 100644
--- a/internal/moves/tablewrite.go
+++ b/internal/moves/tablewrite.go
@@ -18,7 +18,9 @@ import (
// request at a time through release.PutOnce, so no copy the code did not see
// is ever in the network. An answer is returned: success, or
// release.ErrCASConflict (our attempt landed or another write did; the
-// caller's next read tells which). A timeout, a transport error or a 5xx is
+// caller's next read tells which), or release.ErrPutRejected (a non-CAS
+// 4xx such as a 403 or a 400: the server refused the write, which failed
+// and is never resent). Only a timeout, a transport error or a 5xx is
// no answer, and the identical request (the same bytes, the same condition)
// is sent again on offlinecleanup.RetrySchedule until it is answered or the
// context ends. A write the context gave up on stays registered in its
@@ -114,6 +116,8 @@ func (tw *TableWrites) Resend(ctx context.Context, s release.Store) {
// PutTable writes body to OfflineActiveKey under ifMatch ("" is create-only,
// as release.Store.Put) and answers the new ETag, release.ErrCASConflict,
+// release.ErrPutRejected wrapped (the server refused it: failed, and its
+// registration is removed like any answered one),
// release.ErrNoSingleAttemptPut (nothing was sent: s has no PutOnce), or,
// when ctx ends before any answer, ErrUncertain wrapped, with the write left
// registered in tw, unowned, for tw.Resend.
@@ -145,9 +149,10 @@ func sendTableWrite(ctx context.Context, s release.Store, w *tableWrite) (string
return release.PutOnce(ctx, s, OfflineActiveKey, w.body, w.ifMatch)
}
-// tableAnswered: success, a lost condition, or a store that sent nothing.
+// tableAnswered: success, a lost condition, a rejection, or a store that
+// sent nothing. Only a timeout, a transport error or a 5xx is unanswered.
func tableAnswered(err error) bool {
- return err == nil || errors.Is(err, release.ErrCASConflict) || errors.Is(err, release.ErrNoSingleAttemptPut)
+ return err == nil || errors.Is(err, release.ErrCASConflict) || errors.Is(err, release.ErrPutRejected) || errors.Is(err, release.ErrNoSingleAttemptPut)
}
// tableRetrySchedule is offlinecleanup.RetrySchedule unless a test set its
diff --git a/internal/moves/tablewrite_test.go b/internal/moves/tablewrite_test.go
index b631768cb..a2ff2a3be 100644
--- a/internal/moves/tablewrite_test.go
+++ b/internal/moves/tablewrite_test.go
@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"errors"
+ "fmt"
"sync"
"testing"
"time"
@@ -90,6 +91,62 @@ func TestUnansweredTableWriteIsResentIdentically(t *testing.T) {
}
}
+// A write the server rejected (a non-CAS 4xx: 403, 400) is an answer:
+// failed, never resent, its registration removed by its owner, whether the
+// owner is PutTable or a Resend. A timeout before it is still resent.
+func TestRejectedTableWriteIsAnAnswer(t *testing.T) {
+ fastResend(t)
+ ctx := context.Background()
+ f := newTableStore()
+ etag := f.set(OfflineActiveKey, tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 1}))
+ forbidden := fmt.Errorf("fake: 403 AccessDenied: %w", release.ErrPutRejected)
+ f.onPutOnce = func(n int, _ string, _ []byte, _ string) (bool, error) {
+ if n == 0 {
+ return false, errTimeout
+ }
+ return false, forbidden
+ }
+ var tw TableWrites
+ body := tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 2})
+ if _, err := PutTable(ctx, f, &tw, body, etag); !errors.Is(err, release.ErrPutRejected) || errors.Is(err, ErrUncertain) {
+ t.Fatalf("a 403 = %v, want the rejection, not uncertain", err)
+ }
+ if f.Requests() != 2 || tw.Unanswered() != 0 {
+ t.Fatalf("after a 403: requests %d, Unanswered %d, want 2 and 0", f.Requests(), tw.Unanswered())
+ }
+
+ // A write left unanswered, then rejected on its Resend, is removed.
+ g := newTableStore()
+ e1 := g.set(OfflineActiveKey, tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 1}))
+ var mu sync.Mutex
+ reject := false
+ g.onPutOnce = func(int, string, []byte, string) (bool, error) {
+ mu.Lock()
+ defer mu.Unlock()
+ if reject {
+ return false, forbidden
+ }
+ return false, errTimeout
+ }
+ short, cancel := context.WithTimeout(ctx, 20*time.Millisecond)
+ defer cancel()
+ if _, err := PutTable(short, g, &tw, body, e1); !errors.Is(err, ErrUncertain) {
+ t.Fatalf("timeouts = %v, want ErrUncertain", err)
+ }
+ mu.Lock()
+ reject = true
+ mu.Unlock()
+ n := g.Requests()
+ tw.Resend(ctx, g)
+ if tw.Unanswered() != 0 || g.Requests() != n+1 {
+ t.Fatalf("a rejected Resend: Unanswered %d, requests %d (want 0, %d)", tw.Unanswered(), g.Requests(), n+1)
+ }
+ tw.Resend(ctx, g)
+ if g.Requests() != n+1 {
+ t.Fatal("a rejected write was sent again")
+ }
+}
+
// Review 8, item 6: a registration is removed only by its owner, after its
// final answer, and ownership changes hands only while no attempt of the
// write is inside PutOnce. Run under -race.
diff --git a/internal/release/putonce_test.go b/internal/release/putonce_test.go
index 8c7307fef..98dc26868 100644
--- a/internal/release/putonce_test.go
+++ b/internal/release/putonce_test.go
@@ -172,3 +172,39 @@ func (c *putCountingStore) Put(ctx context.Context, key string, body []byte, ifM
c.puts++
return c.Store.Put(ctx, key, body, ifMatch)
}
+
+// A non-CAS 4xx is the server's rejection of the write: an answer, wrapped
+// as ErrPutRejected, after one request. A timeout-like 4xx (408, 429), a
+// missing bucket's 404 and a 5xx are not answers.
+func TestPutOnceRejectionIsAnAnswer(t *testing.T) {
+ ctx := context.Background()
+ cases := []struct {
+ status int
+ code string
+ ifMatch string
+ rejected bool
+ cas bool
+ }{
+ {http.StatusForbidden, "AccessDenied", `"e1"`, true, false},
+ {http.StatusBadRequest, "InvalidRequest", `"e1"`, true, false},
+ {http.StatusNotFound, "NoSuchKey", "", true, false},
+ {http.StatusNotFound, "NoSuchKey", `"e1"`, false, true},
+ {http.StatusNotFound, "NoSuchBucket", "", false, false},
+ {http.StatusRequestTimeout, "RequestTimeout", `"e1"`, false, false},
+ {http.StatusTooManyRequests, "SlowDown", `"e1"`, false, false},
+ {http.StatusServiceUnavailable, "SlowDown", `"e1"`, false, false},
+ {http.StatusPreconditionFailed, "PreconditionFailed", `"e1"`, false, true},
+ }
+ for _, c := range cases {
+ srv := &putCounter{status: c.status, code: c.code}
+ h := httptest.NewServer(srv)
+ _, err := defaultRetryS3(h.URL).PutOnce(ctx, "control/key.json", []byte("{}"), c.ifMatch)
+ h.Close()
+ if err == nil || errors.Is(err, ErrPutRejected) != c.rejected || errors.Is(err, ErrCASConflict) != c.cas {
+ t.Errorf("%d %s (If-Match %q): %v, want rejected %v, conflict %v", c.status, c.code, c.ifMatch, err, c.rejected, c.cas)
+ }
+ if n := srv.count(); n != 1 {
+ t.Errorf("%d %s: %d requests, want 1", c.status, c.code, n)
+ }
+ }
+}
diff --git a/internal/release/store.go b/internal/release/store.go
index 3e3bf1f16..6f980d969 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -65,6 +65,12 @@ type SingleAttemptPutter interface {
PutOnce(ctx context.Context, key string, body []byte, ifMatch string) (string, error)
}
+// ErrPutRejected is a single attempt the server refused outright: a 4xx
+// that is neither a lost condition (ErrCASConflict) nor transient (408,
+// 429, a missing bucket's 404). The write was not applied and resending it
+// cannot change that, so it is an answer, not an uncertainty.
+var ErrPutRejected = errors.New("store: put rejected")
+
// ErrNoSingleAttemptPut is a store that cannot send a single attempt (the
// store door's client, whose door writes through a retrying client):
// nothing was sent.
@@ -799,6 +805,26 @@ func s3CASConflict(err error) bool {
return false
}
+// s3Rejected reports whether a write's error is the server refusing it: a
+// 4xx other than 408 (a timeout), 429 (throttling) and a missing bucket's
+// 404, which can pass. A transport error carries no status and is never a
+// rejection; neither is a 5xx.
+func s3Rejected(err error) bool {
+ var httpErr interface{ HTTPStatusCode() int }
+ if !errors.As(err, &httpErr) {
+ return false
+ }
+ code := httpErr.HTTPStatusCode()
+ if code < 400 || code > 499 || code == http.StatusRequestTimeout || code == http.StatusTooManyRequests {
+ return false
+ }
+ var apiErr smithy.APIError
+ if code == http.StatusNotFound && errors.As(err, &apiErr) && apiErr.ErrorCode() == "NoSuchBucket" {
+ return false
+ }
+ return true
+}
+
func (s *S3Store) Get(ctx context.Context, key string) ([]byte, string, error) {
out, err := s.Client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key})
if err != nil {
@@ -882,6 +908,9 @@ func (s *S3Store) PutOnce(ctx context.Context, key string, body []byte, ifMatch
if s3CASConflict(err) || (ifMatch != "" && s3NotFound(err)) {
return "", fmt.Errorf("s3store: %s: %w", key, ErrCASConflict)
}
+ if s3Rejected(err) {
+ return "", fmt.Errorf("s3store: %s: %w: %w", key, ErrPutRejected, err)
+ }
return "", err
}
return aws.ToString(out.ETag), nil