diff --git a/README.md b/README.md index 9c1041c7..461b030f 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,7 @@ When configuring S3 clients, ensure path-style addressing is enabled. See [docs/ - Set `block_caching_enabled: false` to cache whole objects instead — a range request then triggers a background fetch of the full object (if within threshold) - In tiered mode, a PUT routed to the local tier checks a concrete SigV4 `X-Amz-Content-Sha256` digest after admission and any local precondition pass and the full body passes its declared-length and any Content-MD5 checks; conditional writes against upstream-tier markers, `UNSIGNED-PAYLOAD`, and aws-chunked uploads keep their existing paths - PUT/DELETE operations invalidate the cache entry +- Legacy ordered metadata populates are refused when the tombstone cannot be read or is malformed, rather than assuming no invalidation. The detached metadata warm remains best-effort, so a successful S3 write response is unchanged. - If a whole-object cache write returns an error during a successful GET, the upstream response is unchanged and the cache-populate slot becomes available after relay cleanup, so a later eligible GET can retry. A store call that never returns has no completion guarantee. See [docs/cache-control.md](docs/cache-control.md) for detailed cache control and revalidation documentation. For tuning and sizing block caching in production, see [docs/deploy.md](docs/deploy.md#block-aligned-caching). diff --git a/cache/coordinator.go b/cache/coordinator.go index 7c3baaac..c33a2ef1 100644 --- a/cache/coordinator.go +++ b/cache/coordinator.go @@ -40,6 +40,8 @@ import ( "github.com/rs/zerolog/log" cacheclient "github.com/tigrisdata/ocache/client" "github.com/tigrisdata/tag/metrics" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // metaCoordinator is the ordering seam for the mutable meta key. @@ -290,8 +292,15 @@ func (c *legacyCoordinator) putMeta(ctx context.Context, bucket, key, metaKey st if expected != VersionAny { // Tombstone gate right before the visibility-granting meta write: an // invalidation at or after the caller's decision-time stamp blocks the - // commit, so a populate cannot resurrect deleted metadata. - if ts := c.tombstoneTimestamp(ctx, bucket, key); ts >= int64(expected) { + // commit, so a populate cannot resurrect deleted metadata. If the marker + // cannot be read or decoded, its state is unknown, not absent: refuse the + // visibility write and let the caller handle this as a cache-write error. + ts, err := c.tombstoneTimestamp(ctx, bucket, key) + if err != nil { + metrics.RecordCacheOperation("meta_put", "error") + return false, err + } + if ts >= int64(expected) { metrics.RecordCacheOperation("meta_put", "precondition_lost") log.Debug().Str("bucket", bucket).Str("key", key). Int64("tombstone_ts", ts).Uint64("write_start", expected). @@ -360,12 +369,25 @@ func (c *legacyCoordinator) writeTombstone(ctx context.Context, bucket, key stri return c.client.Put(ctx, MakeTombstoneKey(bucket, key), data, c.tombstoneTTL) } -func (c *legacyCoordinator) tombstoneTimestamp(ctx context.Context, bucket, key string) int64 { +func (c *legacyCoordinator) tombstoneTimestamp(ctx context.Context, bucket, key string) (int64, error) { data, err := c.client.Get(ctx, MakeTombstoneKey(bucket, key)) - if err != nil || len(data) != 8 { - return 0 + if err != nil { + // The production embedded client represents absence as nil, nil; the + // gRPC and in-memory clients return a typed NotFound status. Other errors + // (including routing errors whose text contains "not found") leave the + // marker state unknown and must not authorize a metadata write. + if data == nil && status.Code(err) == codes.NotFound { + return 0, nil + } + return 0, fmt.Errorf("read tombstone for %s/%s: %w", bucket, key, err) + } + if data == nil { + return 0, nil + } + if len(data) != 8 { + return 0, fmt.Errorf("malformed tombstone for %s/%s: got %d bytes, want 8", bucket, key, len(data)) } - return int64(binary.BigEndian.Uint64(data)) + return int64(binary.BigEndian.Uint64(data)), nil } // deleteMetaIfVersion is a CAS-strength identity delete: legacy coordination diff --git a/cache/coordinator_tombstone_test.go b/cache/coordinator_tombstone_test.go new file mode 100644 index 00000000..95921761 --- /dev/null +++ b/cache/coordinator_tombstone_test.go @@ -0,0 +1,160 @@ +package cache + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" + cacheclient "github.com/tigrisdata/ocache/client" + ocachecoordinator "github.com/tigrisdata/ocache/coordinator" + "github.com/tigrisdata/tag/config" + "github.com/tigrisdata/tag/metrics" +) + +type tombstoneGetOverrideClient struct { + cacheclient.CacheClient + markerKey string + data []byte + err error + override bool +} + +func (c *tombstoneGetOverrideClient) Get(ctx context.Context, key string) ([]byte, error) { + if c.override && key == c.markerKey { + return c.data, c.err + } + return c.CacheClient.Get(ctx, key) +} + +func TestLegacyCoordinator_TombstoneReadErrorAndMalformedMarkerRefuseMetaWrite(t *testing.T) { + const bucket, key = "b", "k" + markerKey := MakeTombstoneKey(bucket, key) + backendErr := errors.New("cache read unavailable") + routingErr := ocachecoordinator.NewNodeNotFoundError("missing-node", markerKey) + tests := []struct { + name string + data []byte + err error + override bool + wantCause error + wantErrorText string + }{ + { + name: "backend read failure", + err: backendErr, + override: true, + wantCause: backendErr, + }, + { + name: "routing error containing not found", + err: routingErr, + override: true, + wantCause: ocachecoordinator.ErrNodeNotFound, + }, + { + name: "empty marker is malformed", + data: make([]byte, 0), + override: true, + wantErrorText: "malformed tombstone", + }, + { + name: "short marker is malformed", + data: []byte{1, 2, 3}, + override: true, + wantErrorText: "malformed tombstone", + }, + { + name: "long marker is malformed", + data: make([]byte, 9), + override: true, + wantErrorText: "malformed tombstone", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg := config.NewDefault() + cfg.Cache.SetLegacyCoordination(true) + client := &tombstoneGetOverrideClient{ + CacheClient: cacheclient.NewMemoryCache(), + markerKey: markerKey, + data: tc.data, + err: tc.err, + override: tc.override, + } + c := NewCacheWithClient(client, &cfg.Cache) + ctx := context.Background() + + old, expected, found, err := c.GetMetaWithVersion(ctx, bucket, key) + if err != nil || found || old != nil || expected == 0 { + t.Fatalf("decision token = (meta=%+v expected=%d found=%v err=%v), want absent metadata and a nonzero token", old, expected, found, err) + } + + meta := &CachedObjectMeta{Bucket: bucket, Key: key, ETag: `"stale"`, StatusCode: 200} + metaPutErrorsBefore := testutil.ToFloat64(metrics.CacheOperations.WithLabelValues("meta_put", "error")) + wrote, writeErr := c.PutMetaIfVersion(ctx, bucket, key, meta, 60, expected) + if got := testutil.ToFloat64(metrics.CacheOperations.WithLabelValues("meta_put", "error")); got != metaPutErrorsBefore+1 { + t.Errorf("meta_put/error count = %v, want one additional error outcome", got) + } + if wrote { + t.Fatal("metadata was published despite an unreadable or malformed tombstone") + } + if writeErr == nil { + t.Fatal("PutMetaIfVersion returned nil error for an unreadable or malformed tombstone") + } + if tc.wantCause != nil && !errors.Is(writeErr, tc.wantCause) { + t.Fatalf("PutMetaIfVersion error %v does not preserve cause %v", writeErr, tc.wantCause) + } + if tc.wantErrorText != "" && !strings.Contains(writeErr.Error(), tc.wantErrorText) { + t.Fatalf("PutMetaIfVersion error = %v, want text %q", writeErr, tc.wantErrorText) + } + if _, found, err := c.GetMeta(ctx, bucket, key); err != nil || found { + t.Fatalf("metadata after refused write: found=%v err=%v, want absent", found, err) + } + }) + } +} + +func TestLegacyCoordinator_TombstoneAbsenceStillAllowsMetaWarm(t *testing.T) { + const bucket, key = "b", "k" + for _, tc := range []struct { + name string + override bool + }{ + {name: "typed not found from cache client"}, + {name: "embedded nil result", override: true}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := config.NewDefault() + cfg.Cache.SetLegacyCoordination(true) + client := &tombstoneGetOverrideClient{ + CacheClient: cacheclient.NewMemoryCache(), + markerKey: MakeTombstoneKey(bucket, key), + override: tc.override, + } + c := NewCacheWithClient(client, &cfg.Cache) + ctx := context.Background() + + _, expected, found, err := c.GetMetaWithVersion(ctx, bucket, key) + if err != nil || found || expected == 0 { + t.Fatalf("decision token = (expected=%d found=%v err=%v), want an absent key and nonzero token", expected, found, err) + } + + meta := &CachedObjectMeta{Bucket: bucket, Key: key, ETag: `"current"`, StatusCode: 200} + metaPutErrorsBefore := testutil.ToFloat64(metrics.CacheOperations.WithLabelValues("meta_put", "error")) + wrote, err := c.PutMetaIfVersion(ctx, bucket, key, meta, 60, expected) + if got := testutil.ToFloat64(metrics.CacheOperations.WithLabelValues("meta_put", "error")); got != metaPutErrorsBefore { + t.Errorf("meta_put/error count = %v, want no additional error outcome for a genuinely absent tombstone", got) + } + if err != nil || !wrote { + t.Fatalf("PutMetaIfVersion on absent tombstone = (%v, %v), want (true, nil)", wrote, err) + } + got, found, err := c.GetMeta(ctx, bucket, key) + if err != nil || !found || got == nil || got.ETag != meta.ETag { + t.Fatalf("metadata after warm = (%+v, %v, %v), want current ETag", got, found, err) + } + }) + } +} diff --git a/docs/configuration.md b/docs/configuration.md index 1fdfda8a..aeb40bad 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -189,11 +189,12 @@ cache: block_caching_enabled: true # Meta invalidation/populate coordination. true (default) = legacy tombstone mechanism, - # compatible with any cluster mix — including nodes upgrading directly from <=v1.20. false = - # fenced-CAS coordination (stronger ordering; requires every node in the cluster to be - # CAS-capable BEFORE flipping, then a brisk rolling restart). Standalone deployments may flip - # immediately. Override with TAG_CACHE_LEGACY_COORDINATION (fail-safe: only an explicit - # "false"/"0" selects CAS). + # compatible with any cluster mix — including nodes upgrading directly from <=v1.20. A missing + # tombstone permits a warm; a failed read or malformed marker refuses an ordered metadata write + # instead of assuming no invalidation. false = fenced-CAS coordination (stronger ordering; + # requires every node in the cluster to be CAS-capable BEFORE flipping, then a brisk rolling + # restart). Standalone deployments may flip immediately. Override with + # TAG_CACHE_LEGACY_COORDINATION (fail-safe: only an explicit "false"/"0" selects CAS). legacy_coordination: true # Block granularity AND the whole-vs-block boundary for every populate path: an object @@ -404,7 +405,7 @@ Controls the embedded cache behavior. TAG uses an embedded OCache instance with | `ttl` | duration | `24h` | Default TTL for cached objects | | `size_threshold` | int64 | `1073741824` | Max object size to cache (bytes) | | `block_caching_enabled` | bool | `true` | Enable block-aligned caching for large objects (RFC 0001) | -| `legacy_coordination` | bool | `true` | Meta coordination mechanism: `true` = legacy tombstones (safe for any cluster mix, incl. direct ≤v1.20 upgrades); `false` = fenced CAS (all nodes must be CAS-capable before flipping) | +| `legacy_coordination` | bool | `true` | Meta coordination mechanism: `true` = legacy tombstones (safe for any cluster mix, incl. direct ≤v1.20 upgrades; unreadable or malformed markers refuse ordered metadata writes); `false` = fenced CAS (all nodes must be CAS-capable before flipping) | | `block_size` | int64 | `1048576` | Block granularity **and** the read-side whole-vs-block boundary (must stay below ocache's 64 MB compaction threshold) | | `parquet_optimization` | bool | `false` | Prefetch a parquet object's metadata blocks on a tail read (only when the metadata spans more than the tail block) | | `disk_path` | string | `/var/cache/tag` | Path to cache data directory | diff --git a/docs/metrics.md b/docs/metrics.md index 3d563f7a..c7e109d4 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -112,10 +112,16 @@ Total number of cache misses. Total number of cache operations. -| Label | Description | -| ----------- | ----------------------------------------- | -| `operation` | Operation type: `get`, `put`, `delete` | -| `result` | Result: `hit`, `miss`, `success`, `error` | +| Label | Description | +| ----------- | ------------------------------------------------------------------------ | +| `operation` | Operation type: `get`, `put`, `meta_put`, `delete` | +| `result` | Result: `hit`, `miss`, `success`, `error`, `precondition_lost` | + +`meta_put` denotes metadata publication outcomes. `precondition_lost` means an +ordered metadata write was refused: in legacy coordination, no decision token was +supplied (`expected == 0`) or a newer tombstone was observed; in CAS coordination, +a `PutIfVersion` version mismatch occurred. With legacy coordination, an unreadable +or malformed tombstone is recorded as `error` and the write is refused. **Example queries:** diff --git a/proxy/meta_on_write_tombstone_race_test.go b/proxy/meta_on_write_tombstone_race_test.go new file mode 100644 index 00000000..14ae1a9d --- /dev/null +++ b/proxy/meta_on_write_tombstone_race_test.go @@ -0,0 +1,266 @@ +package proxy + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + cacheclient "github.com/tigrisdata/ocache/client" + "github.com/tigrisdata/tag/cache" + "github.com/tigrisdata/tag/config" +) + +const ( + metaRaceBucket = "meta-race-bucket" + metaRaceKey = "meta-race-key" + metaRaceOldETag = `"v1"` + metaRaceNewETag = `"v2"` +) + +var errTombstoneReadInjected = errors.New("injected tombstone read failure") + +type tombstoneReadFaultClient struct { + cacheclient.CacheClient + markerKey string + armed atomic.Bool + failed chan struct{} +} + +func (c *tombstoneReadFaultClient) Get(ctx context.Context, key string) ([]byte, error) { + if key == c.markerKey && c.armed.CompareAndSwap(true, false) { + close(c.failed) + return nil, errTombstoneReadInjected + } + return c.CacheClient.Get(ctx, key) +} + +type multipartTombstoneRaceForwarder struct { + mockForwarder + + mu sync.Mutex + currentETag string + headCalls int + firstHeadReady chan string + releaseFirstHead <-chan struct{} +} + +func (f *multipartTombstoneRaceForwarder) ForwardWithCapture(_ context.Context, w http.ResponseWriter, r *http.Request) (*ResponseCapture, error) { + etag := metaRaceOldETag + if r.URL.Query().Get("uploadId") == "second" { + etag = metaRaceNewETag + } + body := []byte(fmt.Sprintf(`%s`, etag)) + + f.mu.Lock() + f.currentETag = etag + f.mu.Unlock() + + w.WriteHeader(http.StatusOK) + if _, err := w.Write(body); err != nil { + return nil, err + } + return &ResponseCapture{ + StatusCode: http.StatusOK, + Headers: http.Header{}, + Body: body, + Complete: true, + }, nil +} + +func (f *multipartTombstoneRaceForwarder) DoConditionalHeadRequest(_ context.Context, _, _, _, _, _ string, _ int64) (*http.Response, error) { + f.mu.Lock() + f.headCalls++ + call := f.headCalls + etag := f.currentETag + f.mu.Unlock() + + resp := multipartHeadResponse(etag) + if call == 1 { + f.firstHeadReady <- etag + <-f.releaseFirstHead + } + return resp, nil +} + +func (f *multipartTombstoneRaceForwarder) Forward(_ context.Context, w http.ResponseWriter, _ *http.Request) error { + f.mu.Lock() + etag := f.currentETag + f.mu.Unlock() + + w.Header().Set("ETag", etag) + w.Header().Set("Content-Length", "8") + w.WriteHeader(http.StatusOK) + return nil +} + +func (f *multipartTombstoneRaceForwarder) DoRequestWithCreds(_ context.Context, r *http.Request, _, _ string) (*http.Response, error) { + f.mu.Lock() + etag := f.currentETag + f.mu.Unlock() + + if r.Header.Get("If-None-Match") == etag { + header := make(http.Header) + header.Set("ETag", etag) + return &http.Response{ + StatusCode: http.StatusNotModified, + Header: header, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + } + + body := []byte("new-body") + header := make(http.Header) + header.Set("ETag", etag) + header.Set("Content-Length", fmt.Sprint(len(body))) + header.Set("Content-Type", "application/octet-stream") + return &http.Response{ + StatusCode: http.StatusOK, + Header: header, + Body: io.NopCloser(bytes.NewReader(body)), + ContentLength: int64(len(body)), + }, nil +} + +func multipartHeadResponse(etag string) *http.Response { + header := make(http.Header) + header.Set("ETag", etag) + header.Set("Content-Length", "8") + header.Set("Content-Type", "application/octet-stream") + header.Set("Last-Modified", time.Unix(1_700_000_000, 0).UTC().Format(http.TimeFormat)) + return &http.Response{ + StatusCode: http.StatusOK, + Header: header, + Body: io.NopCloser(bytes.NewReader(nil)), + ContentLength: 8, + } +} + +func completeMultipartRaceRequest(uploadID string) *http.Request { + return httptest.NewRequest(http.MethodPost, "/"+metaRaceBucket+"/"+metaRaceKey+"?uploadId="+uploadID, nil) +} + +// TestCompleteMultipart_TombstoneReadFailureDoesNotResurrectOldMetadata holds the +// first write's metadata HEAD after it has observed the old ETag, completes a newer +// upload on the same key, then fails only the first worker's tombstone read at commit. +func TestCompleteMultipart_TombstoneReadFailureDoesNotResurrectOldMetadata(t *testing.T) { + cfg := config.NewDefault() + cfg.Cache.SetLegacyCoordination(true) + cfg.Cache.SetBlockCachingEnabled(true) + cfg.Cache.BlockSize = 4 + cfg.Cache.SizeThreshold = 1 << 20 + cfg.Cache.MetaOnWrite = true + cfg.Cache.WarmOnWrite = false + cfg.Cache.ParquetOptimization = false + + faultyClient := &tombstoneReadFaultClient{ + CacheClient: cacheclient.NewMemoryCache(), + markerKey: cache.MakeTombstoneKey(metaRaceBucket, metaRaceKey), + failed: make(chan struct{}), + } + store := cache.NewCacheWithClient(faultyClient, &cfg.Cache) + releaseFirstHead := make(chan struct{}) + var releaseOnce sync.Once + releaseHead := func() { releaseOnce.Do(func() { close(releaseFirstHead) }) } + defer releaseHead() + + forwarder := &multipartTombstoneRaceForwarder{ + firstHeadReady: make(chan string, 1), + releaseFirstHead: releaseFirstHead, + } + svc := NewService(forwarder, store, cfg) + + firstResponse := httptest.NewRecorder() + if err := svc.HandleCompleteMultipartUpload(firstResponse, completeMultipartRaceRequest("first")); err != nil { + t.Fatalf("first completion: %v", err) + } + if firstResponse.Code != http.StatusOK || firstResponse.Body.String() != ``+metaRaceOldETag+`` { + t.Fatalf("first completion response = %d %q, want successful v1 XML", firstResponse.Code, firstResponse.Body.String()) + } + + select { + case etag := <-forwarder.firstHeadReady: + if etag != metaRaceOldETag { + t.Fatalf("first HEAD observed ETag %q, want %q", etag, metaRaceOldETag) + } + case <-time.After(3 * time.Second): + t.Fatal("first meta-on-write HEAD did not start") + } + + secondResponse := httptest.NewRecorder() + if err := svc.HandleCompleteMultipartUpload(secondResponse, completeMultipartRaceRequest("second")); err != nil { + t.Fatalf("second completion: %v", err) + } + if secondResponse.Code != http.StatusOK || secondResponse.Body.String() != ``+metaRaceNewETag+`` { + t.Fatalf("second completion response = %d %q, want successful v2 XML", secondResponse.Code, secondResponse.Body.String()) + } + + const dedupKey = "meta:" + metaRaceBucket + "/" + metaRaceKey + if _, active := svc.activeBackgroundFetches.Load(dedupKey); !active { + t.Fatal("first meta warm left its per-key marker while its HEAD was held") + } + forwarder.mu.Lock() + headCalls := forwarder.headCalls + currentETag := forwarder.currentETag + forwarder.mu.Unlock() + if headCalls != 1 || currentETag != metaRaceNewETag { + t.Fatalf("after second completion, HEAD calls=%d current ETag=%q; want one held old HEAD and current v2", headCalls, currentETag) + } + + faultyClient.armed.Store(true) + releaseHead() + select { + case <-faultyClient.failed: + case <-time.After(3 * time.Second): + t.Fatal("the first worker did not read the tombstone after its HEAD was released") + } + waitForBackgroundMetaWarm(t, svc, dedupKey) + + meta, found, err := store.GetMeta(context.Background(), metaRaceBucket, metaRaceKey) + if err != nil { + t.Fatalf("read metadata after both completions: %v", err) + } + oldMetaVisible := found && meta != nil && meta.ETag == metaRaceOldETag + + headRecorder := httptest.NewRecorder() + headRequest := httptest.NewRequest(http.MethodHead, "/"+metaRaceBucket+"/"+metaRaceKey, nil) + if err := svc.HandleHeadObject(headRecorder, headRequest); err != nil { + t.Fatalf("HEAD after both completions: %v", err) + } + oldMetaFromHead := headRecorder.Header().Get("ETag") == metaRaceOldETag + + getRecorder := httptest.NewRecorder() + getRequest := httptest.NewRequest(http.MethodGet, "/"+metaRaceBucket+"/"+metaRaceKey, nil) + getRequest.Header.Set("If-None-Match", metaRaceOldETag) + if err := svc.HandleGetObject(getRecorder, getRequest); err != nil { + t.Fatalf("conditional GET after both completions: %v", err) + } + oldMetaFromGet := getRecorder.Code == http.StatusNotModified || getRecorder.Header().Get("ETag") == metaRaceOldETag + + if oldMetaVisible || oldMetaFromHead || oldMetaFromGet { + t.Errorf("older multipart metadata remained visible after a newer completion: stored=%v HEAD=(%d,%q) conditional GET=(%d,%q)", + oldMetaVisible, headRecorder.Code, headRecorder.Header().Get("ETag"), getRecorder.Code, getRecorder.Header().Get("ETag")) + } +} + +func waitForBackgroundMetaWarm(t *testing.T, svc *Service, key string) { + t.Helper() + timeout := time.After(3 * time.Second) + for { + if _, active := svc.activeBackgroundFetches.Load(key); !active { + return + } + select { + case <-timeout: + t.Fatal("first meta-on-write worker did not finish after its tombstone read") + case <-time.After(time.Millisecond): + } + } +}