From 25edfad56c72d7a7bc56daab8dac642c45faaa03 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Sat, 10 Oct 2026 05:28:51 +0000 Subject: [PATCH 1/3] test: cover delayed tiered re-tier after local put Signed-off-by: Perfloop Agent --- proxy/tiered_test.go | 353 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 351 insertions(+), 2 deletions(-) diff --git a/proxy/tiered_test.go b/proxy/tiered_test.go index c3d5e8ae..1b26da24 100644 --- a/proxy/tiered_test.go +++ b/proxy/tiered_test.go @@ -10,6 +10,7 @@ import ( "io" "net/http" "net/http/httptest" + "strconv" "strings" "sync" "sync/atomic" @@ -24,6 +25,12 @@ import ( // newTieredTestService builds a Service in tiered mode with an in-memory cache // and the given small/large tier boundary. func newTieredTestService(forwarder RequestForwarder, threshold int64) (*Service, *cache.Cache) { + return newTieredTestServiceWithClient(forwarder, threshold, cacheclient.NewMemoryCache()) +} + +// newTieredTestServiceWithClient keeps the production tiered configuration and +// CAS coordinator while allowing tests to gate versioned cache reads. +func newTieredTestServiceWithClient(forwarder RequestForwarder, threshold int64, client cacheclient.CacheClient) (*Service, *cache.Cache) { cfg := config.NewDefault() cfg.Mode = config.ModeTiered cfg.Cache.SetBlockCachingEnabled(false) @@ -36,8 +43,7 @@ func newTieredTestService(forwarder RequestForwarder, threshold int64) (*Service panic(err) } - memCache := cacheclient.NewMemoryCache() - c := cache.NewCacheWithClient(memCache, &cfg.Cache) + c := cache.NewCacheWithClient(client, &cfg.Cache) svc := NewService(forwarder, c, cfg) return svc, c } @@ -460,6 +466,349 @@ func waitRetierDone(t *testing.T, svc *Service, bucket, key string) { } } +type tieredMetaReadObservation struct { + data []byte + version uint64 + found bool + err error +} + +// tieredMetaReadGateClient pauses a versioned metadata result after the cache +// read returns, so a test can deterministically place a write before GET-side +// re-tier registration. +type tieredMetaReadGateClient struct { + cacheclient.CacheClient + mu sync.Mutex + + key string + blockNext bool + observed chan tieredMetaReadObservation + release <-chan struct{} + watchNext bool + watched chan tieredMetaReadObservation +} + +func (g *tieredMetaReadGateClient) gateNextMetaRead(key string) (<-chan tieredMetaReadObservation, chan struct{}) { + g.mu.Lock() + defer g.mu.Unlock() + g.key = key + g.blockNext = true + g.observed = make(chan tieredMetaReadObservation, 1) + release := make(chan struct{}) + g.release = release + return g.observed, release +} + +func (g *tieredMetaReadGateClient) watchNextMetaRead() <-chan tieredMetaReadObservation { + g.mu.Lock() + defer g.mu.Unlock() + g.watchNext = true + g.watched = make(chan tieredMetaReadObservation, 1) + return g.watched +} + +func (g *tieredMetaReadGateClient) GetWithVersion(ctx context.Context, key string) ([]byte, uint64, bool, error) { + data, version, found, err := g.CacheClient.GetWithVersion(ctx, key) + observation := tieredMetaReadObservation{data: data, version: version, found: found, err: err} + + g.mu.Lock() + if key != g.key { + g.mu.Unlock() + return data, version, found, err + } + if g.blockNext { + g.blockNext = false + observed, release := g.observed, g.release + g.mu.Unlock() + observed <- observation + select { + case <-release: + case <-ctx.Done(): + return nil, 0, false, ctx.Err() + } + return data, version, found, err + } + var watched chan tieredMetaReadObservation + if g.watchNext { + g.watchNext = false + watched = g.watched + } + g.mu.Unlock() + if watched != nil { + watched <- observation + } + return data, version, found, err +} + +type tieredRetierDelayedScenario struct { + name string + localBody string + markerETag string + fetchStatus int + fetchETag string + fetchBody string +} + +func tieredETag(body string) string { + sum := md5.Sum([]byte(body)) + return `"` + hex.EncodeToString(sum[:]) + `"` +} + +func TestTieredRetierDelayedRegistrationPreservesLocalPut(t *testing.T) { + scenarios := []tieredRetierDelayedScenario{ + { + name: "upstream 404 cannot remove local successor", + localBody: "local-after-404", + markerETag: `"upstream-a"`, + fetchStatus: http.StatusNotFound, + }, + { + name: "changed response cannot relabel same-content local successor", + localBody: "same-content", + markerETag: tieredETag("same-content"), + fetchStatus: http.StatusOK, + fetchETag: `"upstream-b"`, + fetchBody: "replacement-body", + }, + } + + var violations []string + setupComplete := true + for _, scenario := range scenarios { + scenario := scenario + var violation string + if !t.Run(scenario.name, func(t *testing.T) { + violation = runTieredRetierDelayedRegistrationScenario(t, scenario) + }) { + setupComplete = false + } + if violation != "" { + violations = append(violations, scenario.name+": "+violation) + } + } + if !setupComplete { + return + } + if len(violations) != 0 { + t.Errorf("stale re-tier changed the authoritative metadata of a completed local PUT: %s", strings.Join(violations, "; ")) + } +} + +func runTieredRetierDelayedRegistrationScenario(t *testing.T, scenario tieredRetierDelayedScenario) string { + t.Helper() + const bucket, key = "b", "obj" + ctx := context.Background() + + mock, _, _ := tieredMock() + cleanupFinished := make(chan struct{}) + mock.doObjectDeleteFunc = func(context.Context, string, string, string, string, string) (*http.Response, error) { + defer close(cleanupFinished) + // The local PUT's upstream cleanup is unrelated to this invariant. A + // lost If-Match takes the normal early-return path before its own + // versioned metadata repair read can be mistaken for the re-tier read. + return &http.Response{StatusCode: http.StatusPreconditionFailed, Body: http.NoBody}, nil + } + var fetches atomic.Int32 + mock.doFullObjectFunc = func(context.Context, string, string, string, string) (*http.Response, error) { + if fetches.Add(1) > 1 { + return nil, errors.New("unexpected repeated re-tier fetch") + } + header := make(http.Header) + if scenario.fetchETag != "" { + header.Set("ETag", scenario.fetchETag) + header.Set("Content-Length", strconv.Itoa(len(scenario.fetchBody))) + } + return &http.Response{ + StatusCode: scenario.fetchStatus, + Header: header, + ContentLength: int64(len(scenario.fetchBody)), + Body: io.NopCloser(strings.NewReader(scenario.fetchBody)), + }, nil + } + + gate := &tieredMetaReadGateClient{CacheClient: cacheclient.NewMemoryCache()} + svc, c := newTieredTestServiceWithClient(mock, 1024, gate) + marker := &cache.CachedObjectMeta{ + Bucket: bucket, Key: key, ETag: scenario.markerETag, + BodyUpstream: true, ContentLength: int64(len(scenario.localBody)), StatusCode: http.StatusOK, + } + _, absentVersion, found, err := c.GetMetaWithVersion(ctx, bucket, key) + if err != nil || found { + t.Fatalf("initial metadata state: found=%v err=%v, want absence", found, err) + } + if wrote, err := c.PutMetaIfVersion(ctx, bucket, key, marker, 60, absentVersion); err != nil || !wrote { + t.Fatalf("seed upstream marker: wrote=%v err=%v", wrote, err) + } + seeded, markerVersion, found, err := c.GetMetaWithVersion(ctx, bucket, key) + if err != nil || !found || seeded == nil || !seeded.BodyUpstream || seeded.ETag != marker.ETag { + t.Fatalf("seeded marker snapshot: meta=%+v found=%v err=%v", seeded, found, err) + } + + observed, release := gate.gateNextMetaRead(cache.MakeMetaKey(bucket, key)) + var releaseOnce sync.Once + resumeGET := func() { releaseOnce.Do(func() { close(release) }) } + getRecorder := httptest.NewRecorder() + getRequest := httptest.NewRequest(http.MethodGet, "/b/obj", nil) + getFinished := make(chan struct{}) + var getErr error + go func() { + getErr = svc.HandleGetObject(getRecorder, getRequest) + close(getFinished) + }() + defer func() { + resumeGET() + select { + case <-getFinished: + case <-time.After(2 * time.Second): + } + }() + + var firstRead tieredMetaReadObservation + select { + case firstRead = <-observed: + case <-time.After(2 * time.Second): + t.Fatal("GET did not reach the gated versioned metadata read") + } + firstMeta, decodeErr := cache.DecodeMeta(firstRead.data) + if firstRead.err != nil || decodeErr != nil || !firstRead.found || firstRead.version != markerVersion || firstMeta == nil || !firstMeta.BodyUpstream || firstMeta.ETag != marker.ETag { + t.Fatalf("gated GET did not observe marker A: meta=%+v version=%d found=%v readErr=%v decodeErr=%v", firstMeta, firstRead.version, firstRead.found, firstRead.err, decodeErr) + } + if svc.retierRunning(bucket, key) { + t.Fatal("re-tier registered before the gated GET resumed") + } + + putRecorder := httptest.NewRecorder() + putRequest := httptest.NewRequest(http.MethodPut, "/b/obj", strings.NewReader(scenario.localBody)) + if err := svc.HandlePutObject(putRecorder, putRequest); err != nil { + t.Fatalf("local successor PUT: %v", err) + } + if putRecorder.Code != http.StatusOK { + t.Fatalf("local successor PUT status = %d, want %d", putRecorder.Code, http.StatusOK) + } + svc.retierMu.Lock() + claims := svc.retierClaims[bucket+"|"+key] + svc.retierMu.Unlock() + if claims != 0 { + t.Fatalf("PUT returned while holding %d re-tier write claims", claims) + } + before, beforeVersion, found, err := c.GetMetaWithVersion(ctx, bucket, key) + if err != nil || !found || before == nil || before.BodyUpstream || before.BodyRef == "" { + t.Fatalf("completed PUT metadata: meta=%+v found=%v err=%v", before, found, err) + } + if beforeVersion == markerVersion { + t.Fatalf("successor metadata version %d did not advance from marker version", beforeVersion) + } + if scenario.localBody == "same-content" && before.ETag != marker.ETag { + t.Fatalf("same-content successor ETag = %q, want marker ETag %q", before.ETag, marker.ETag) + } + select { + case <-cleanupFinished: + case <-time.After(2 * time.Second): + t.Fatal("local PUT's cross-tier cleanup did not finish") + } + + workerRead := gate.watchNextMetaRead() + resumeGET() + select { + case <-getFinished: + case <-time.After(2 * time.Second): + t.Fatal("GET did not finish after its metadata snapshot was released") + } + if getErr != nil { + t.Fatalf("validated GET: %v", getErr) + } + var workerSnapshot tieredMetaReadObservation + select { + case workerSnapshot = <-workerRead: + case <-time.After(2 * time.Second): + t.Fatal("delayed GET did not start the re-tier worker's pre-fetch metadata read") + } + workerMeta, workerDecodeErr := cache.DecodeMeta(workerSnapshot.data) + if workerSnapshot.err != nil || workerDecodeErr != nil || !workerSnapshot.found || workerSnapshot.version != beforeVersion || workerMeta == nil || workerMeta.BodyUpstream || workerMeta.BodyRef != before.BodyRef { + t.Fatalf("re-tier worker did not observe completed local PUT: meta=%+v version=%d found=%v readErr=%v decodeErr=%v", workerMeta, workerSnapshot.version, workerSnapshot.found, workerSnapshot.err, workerDecodeErr) + } + waitRetierDone(t, svc, bucket, key) + + issues := make([]string, 0, 5) + after, afterVersion, found, err := c.GetMetaWithVersion(ctx, bucket, key) + if err != nil { + t.Fatalf("metadata after re-tier: %v", err) + } + if !found || after == nil { + issues = append(issues, "successor metadata was removed") + } else { + if afterVersion != beforeVersion { + issues = append(issues, "metadata version changed") + } + if after.BodyRef != before.BodyRef { + issues = append(issues, "BodyRef changed") + } + if after.ETag != before.ETag { + issues = append(issues, "ETag changed") + } + if after.BodyUpstream { + issues = append(issues, "metadata no longer identifies the local tier") + } + } + + getAfter := httptest.NewRecorder() + if err := svc.HandleGetObject(getAfter, httptest.NewRequest(http.MethodGet, "/b/obj", nil)); err != nil { + t.Fatalf("GET after re-tier: %v", err) + } + waitRetierDone(t, svc, bucket, key) + if getAfter.Code != http.StatusOK || getAfter.Body.String() != scenario.localBody || getAfter.Header().Get("ETag") != before.ETag { + issues = append(issues, "subsequent validated GET did not expose the local PUT") + } + headAfter := httptest.NewRecorder() + if err := svc.HandleHeadObject(headAfter, httptest.NewRequest(http.MethodHead, "/b/obj", nil)); err != nil { + t.Fatalf("HEAD after re-tier: %v", err) + } + if headAfter.Code != http.StatusOK || headAfter.Header().Get("ETag") != before.ETag || headAfter.Header().Get("Content-Length") != strconv.Itoa(len(scenario.localBody)) { + issues = append(issues, "subsequent validated HEAD did not expose the local PUT") + } + return strings.Join(issues, ", ") +} + +// A 404 for an unchanged marker still removes that marker from the +// authoritative view; the stale-GET guard must not disable this convergence. +func TestTieredRetier404ConvergesUnchangedMarker(t *testing.T) { + mock, _, _ := tieredMock() + var fetches atomic.Int32 + mock.doFullObjectFunc = func(context.Context, string, string, string, string) (*http.Response, error) { + fetches.Add(1) + return &http.Response{StatusCode: http.StatusNotFound, Header: make(http.Header), Body: http.NoBody}, nil + } + svc, c := newTieredTestService(mock, 1024) + marker := &cache.CachedObjectMeta{ + Bucket: "b", Key: "obj", ETag: `"gone"`, BodyUpstream: true, + ContentLength: 4, StatusCode: http.StatusOK, + } + _, version, found, err := c.GetMetaWithVersion(context.Background(), "b", "obj") + if err != nil || found { + t.Fatalf("initial metadata state: found=%v err=%v, want absence", found, err) + } + if wrote, err := c.PutMetaIfVersion(context.Background(), "b", "obj", marker, 60, version); err != nil || !wrote { + t.Fatalf("seed upstream marker: wrote=%v err=%v", wrote, err) + } + if w := tieredDo(t, svc, http.MethodGet, "/b/obj", "", nil); w.Code != http.StatusOK { + t.Fatalf("GET status = %d, want forwarded 200", w.Code) + } + waitRetierDone(t, svc, "b", "obj") + if n := fetches.Load(); n != 1 { + t.Fatalf("background fetches = %d, want 1", n) + } + if _, found, err := c.GetMeta(context.Background(), "b", "obj"); err != nil || found { + t.Fatalf("unchanged 404 marker survived convergence: found=%v err=%v", found, err) + } + head := httptest.NewRecorder() + if err := svc.HandleHeadObject(head, httptest.NewRequest(http.MethodHead, "/b/obj", nil)); err != nil { + t.Fatalf("HEAD after 404 convergence: %v", err) + } + if head.Code != http.StatusNotFound { + t.Fatalf("HEAD after 404 convergence = %d, want %d", head.Code, http.StatusNotFound) + } +} + // A small object that landed upstream-tier (a cold-start PUT forwarded before // its key was learned) is healed by the first validated read: the body moves // into the local tier and subsequent reads stop forwarding. From 895aabd44340384d7722e8d54bf60f750c70fdba Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Sat, 10 Oct 2026 05:36:20 +0000 Subject: [PATCH 2/3] fix: fence tiered re-tier writes with trigger version Signed-off-by: Perfloop Agent --- docs/tiered-mode.md | 6 ++-- proxy/tiered.go | 68 ++++++++++++++++++++++++--------------------- 2 files changed, 41 insertions(+), 33 deletions(-) diff --git a/docs/tiered-mode.md b/docs/tiered-mode.md index b2d2304c..900d7b17 100644 --- a/docs/tiered-mode.md +++ b/docs/tiered-mode.md @@ -41,8 +41,10 @@ the body into the local tier — streamed like the engine's PUT, so a heal costs fixed memory regardless of object size and the threshold is never budget-capped. This heals objects mis-placed by the cold-start window below, capping the damage at one extra upstream fetch per object -instead of one body forward per read until TTL. The displaced upstream copy -is left for the upstream bucket's own expiry. +instead of one body forward per read until TTL. Each background move is bound +to the marker version that triggered it; if a later write replaces that +metadata, the stale move cannot change the newer entry. The displaced +upstream copy is left for the upstream bucket's own expiry. ## Authentication diff --git a/proxy/tiered.go b/proxy/tiered.go index 66d54c16..a7436069 100644 --- a/proxy/tiered.go +++ b/proxy/tiered.go @@ -120,7 +120,7 @@ func (s *Service) handleTieredObject(w http.ResponseWriter, r *http.Request) err // forwards through this branch too, and healing on it would launch a // full-body download the triggering request never serves. if r.Method == http.MethodGet && originlessPlainObject(r) { - s.maybeRetierOnRead(bucket, key, accessKey, secretKey, meta) + s.maybeRetierOnRead(bucket, key, accessKey, secretKey, meta, metaVersion) } // The mode's principal body traffic — record it (upstream-sourced), // or a forward-heavy tiered node reads as ~100% local. @@ -213,8 +213,9 @@ func (s *Service) handleTieredPut(w http.ResponseWriter, r *http.Request) error // the in-flight PUT serve the prior version, which is the atomic-replace // behavior clients expect. The one read-triggered populate in this mode — // the re-tier — cannot be ordered against this path's writes here (it - // performs none pre-forward); it defends itself with a claim plus its own - // pre-fetch decision token instead (see maybeRetierOnRead). + // performs none pre-forward). Its claim cancels registered work and its + // triggering GET's version token rejects any stale worker mutation (see + // maybeRetierOnRead). // No post-success invalidation either: the marker overwrites the prior // metadata directly (a displaced local body ages out by TTL, the engine's // own overwrite semantics), which lets the marker commit under the @@ -496,33 +497,36 @@ const ( // that was forwarded before its key was learned. A background fetch moves the // body into the local tier so subsequent reads stop paying a body forward. // +// The triggering GET's metadata version travels with the marker. Before any +// upstream work, the worker verifies that this exact upstream marker is still +// current; every later metadata mutation uses that same version, so a write +// that wins before registration or during the fetch remains authoritative. +// // Guards, in order: // - dedup: one re-tier in flight per key, others ride the existing marker; // - budget: STREAMED like the engine's PUT — the body goes straight into // the cache under a BodyRef, so the reservation is the fixed stream // weight, never the object; shed non-blocking; +// - snapshot: the current version and marker identity must still match the +// triggering GET before the upstream fetch begins; // - version: the fetched ETag must match the marker's — anything else means -// a concurrent write replaced the object, whose state must be left alone; +// a concurrent upstream replacement, whose metadata may be converged only +// while the original marker version is still current; // - identity commit: the entry is re-written only if it still IS the marker -// (same ETag, still upstream-tier); +// (same ETag, still upstream-tier), using the triggering GET's version; // - write claim: PUTs write no tombstones, so a concurrent PUT could land // between the identity check and the commit and be overwritten by the // older version — a local-tier PUT's ONLY copy, in the worst case. Every // tiered PUT therefore holds a per-key claim for its whole duration -// (claimRetierWrite): taking it cancels the running re-tier, and while -// held no new re-tier can register — a GET mid-PUT sees the old marker -// but cannot start a heal against it. Claim and registration share one -// mutex, so every interleaving lands on cancel-or-refuse. What remains is -// two truly concurrent commits racing in the store — the same unordered -// outcome S3 itself gives two concurrent writers; -// - decision tokens: the commit's token is read BEFORE the fetch, so a -// DELETE racing the re-tier enters the store provably after it and the -// commit is refused (DELETEs need no cancellation — the coordinator's -// ordering already covers them). +// (claimRetierWrite): taking it cancels registered work, and while held no +// new re-tier can register. The claim alone cannot stop a GET that read an +// older marker before the claim and registers after it is released; the +// carried version rejects that stale worker. The cache's CAS operation +// protects the final mutation if a write wins after the worker's snapshot. // // The upstream copy is left as an orphan for the upstream bucket's expiry — // deleting it here could race a concurrent write of the same key. -func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, marker *cache.CachedObjectMeta) { +func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, marker *cache.CachedObjectMeta, markerVersion uint64) { if marker.ContentLength < 0 || marker.ContentLength > s.config.Cache.SizeThreshold { return } @@ -561,11 +565,13 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma defer s.releaseCacheSlot(weight) defer cancel() - // Decision token BEFORE the fetch: a DELETE (or any write) racing this - // populate enters the store after this token and provably refuses the - // commit below — the token-model form of the old stamp-before-fetch. - _, decToken, tfound, terr := s.cache.GetMetaWithVersion(ctx, bucket, key) - if terr != nil || !tfound { + // Verify the decision-time snapshot captured by the triggering GET + // before doing upstream work. A write may have completed after that + // GET read the marker but before this worker registered; it must not + // lend its newer version token to work based on the older marker. + current, currentVersion, tfound, terr := s.cache.GetMetaWithVersion(ctx, bucket, key) + if terr != nil || !tfound || current == nil || currentVersion != markerVersion || + !current.BodyUpstream || current.ETag != marker.ETag || current.ContentLength != marker.ContentLength { metrics.RecordTieredRetier("changed") return } @@ -598,8 +604,8 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma if s.retierRecentMismatch != nil { s.retierRecentMismatch.Add(bucket+"|"+key+"|"+etag, struct{}{}) } - // The marker is now PROVEN wrong, and the pre-fetch token is - // still in hand — converge it instead of leaving the + // The marker is now PROVEN wrong, and the triggering GET's + // version is still in hand — converge it instead of leaving the // divergence authoritative for the marker's TTL. Gone (404 — // e.g. the cache bucket's own expiry collected the body): // remove the marker under the token, so HEAD stops answering @@ -610,7 +616,7 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma // refused or failed converge just leaves the backoff doing // its job until the next attempt. if resp.StatusCode == http.StatusNotFound { - if _, derr := s.cache.DeleteMetaIfVersion(ctx, bucket, key, decToken); derr != nil { + if _, derr := s.cache.DeleteMetaIfVersion(ctx, bucket, key, markerVersion); derr != nil { log.Debug().Err(derr).Str("bucket", bucket).Str("key", key).Msg("Re-tier converge: marker removal failed") } } else { @@ -625,7 +631,7 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma if fresh.ContentLength < 0 && drainErr == nil { fresh.ContentLength = drained } - if _, perr := s.cache.PutMetaIfVersion(ctx, bucket, key, fresh, int(s.config.Cache.TTL.Seconds()), decToken); perr != nil { + if _, perr := s.cache.PutMetaIfVersion(ctx, bucket, key, fresh, int(s.config.Cache.TTL.Seconds()), markerVersion); perr != nil { log.Debug().Err(perr).Str("bucket", bucket).Str("key", key).Msg("Re-tier converge: marker rewrite failed") } } @@ -684,9 +690,9 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma // Identity-guarded commit: only re-write the entry if it still IS the // marker this re-tier was triggered by. The commit's expectation is - // the PRE-FETCH decision token, so anything landing after that - // instant — including during the fetch — refuses the commit - // atomically; this re-read is purely the identity guard. The + // the triggering GET's version, so anything landing after that + // snapshot — including before registration or during the fetch — + // refuses the commit atomically; this re-read is the identity guard. The // cancellation check comes after it — a racing PUT cancels before it // stores, so an alive context here means no PUT has entered the store // ahead of us; the token precondition then holds the line for @@ -716,7 +722,7 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma // to copy. meta.ContentLength = src.n ttl := int(s.config.Cache.TTL.Seconds()) - wrote, werr := s.cache.PutMetaIfVersion(ctx, bucket, key, meta, ttl, decToken) + wrote, werr := s.cache.PutMetaIfVersion(ctx, bucket, key, meta, ttl, markerVersion) if werr != nil { // AMBIGUOUS commit (see the engine PUT's rule): the owner may // have applied it, so the staged body must not be deleted — @@ -726,8 +732,8 @@ func (s *Service) maybeRetierOnRead(bucket, key, accessKey, secretKey string, ma return } if !wrote { - // A racer entered the store after the pre-fetch token: the object - // stays on its current state (marker, newer write, or deleted) — + // A racer entered the store after the triggering GET's snapshot: + // the object stays on its current state (marker, newer write, or deleted) — // not a re-tier. Definitive refusal: reclaim the staged body. discardRef() metrics.RecordTieredRetier("changed") From 48f60834fcce578902328f222f07023cd76cd5e2 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Sat, 10 Oct 2026 06:01:27 +0000 Subject: [PATCH 3/3] docs: describe tiered write metadata authority Signed-off-by: Perfloop Agent --- README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/README.md b/README.md index c8d6fdb7..2a772a88 100644 --- a/README.md +++ b/README.md @@ -155,6 +155,10 @@ When multiple concurrent requests arrive for the same uncached object: See [docs/architecture.md](docs/architecture.md) for detailed architecture documentation. +### Tiered Store Mode + +In tiered mode, validated plain-object PUTs no larger than `cache.size_threshold` are stored locally, and their metadata is authoritative. A background re-tier started by a GET is bound to the marker version seen by that GET, so it cannot replace metadata committed by a later write. See [docs/tiered-mode.md](docs/tiered-mode.md) for the full behavior. + ## Performance A single TAG node saturates a 100 Gbps NIC at **~85+ Gbps** for objects 1 MiB and larger, serves **~75K ops/sec** for small objects at sub-millisecond p50, and holds low single-digit-millisecond TTFB — all while using around **12% of available CPU**.