diff --git a/design-bucket-config-serialization.md b/design-bucket-config-serialization.md new file mode 100644 index 000000000..aee383d3e --- /dev/null +++ b/design-bucket-config-serialization.md @@ -0,0 +1,167 @@ +# Design: Serializing Bucket Configuration Mutations + +Issue #9651 — concurrent `PutBucketVersioning` + `PutBucketEncryption` (as Terraform +issues them in parallel) intermittently lose the encryption write. + +## Root cause + +The bucket's entire config lives in one filer entry, `/buckets/`. Every +config API does a read-modify-write of that single entry, and the writes are not +serialized: + +- `updateBucketConfig(bucket, fn)` (`s3api_bucket_config.go:468`) — sources from a + possibly-stale cached `BucketConfig`, mutates `Entry.Extended`, writes the + **whole** entry. Used by: versioning, object-lock config, lifecycle, ACL/owner. +- `UpdateBucketMetadata` → `setBucketMetadata` (`:1042`) — reads a fresh entry, + mutates `Entry.Content`, writes the **whole** entry. Used by: encryption, CORS, + tagging, ownership, policy, notification. + +Two ingredients produce the lost update: + +1. **No serialization** of the read→modify→write (the cache mutexes only guard the + in-memory map, not the RMW). +2. **Whole-entry rewrite from an independent snapshot** — `updateBucketConfig` + rebuilds from a stale cached `BucketConfig` whose `Content` predates the + concurrent encryption write, so writing the whole entry reverts `Content`. + +Sequential calls always pass (each sees the previous write), so it only surfaces +under concurrency — and CI's slower IO widens the window (the "2 of ~12 runs"). + +## Goals + +- No lost updates across concurrent bucket-config changes — for **all** config + fields, not just versioning/encryption. +- Correct for a single S3 gateway (the reported case) and for multiple gateways. +- Reuse the filer primitives just merged (per-path lock, `WriteCondition`, + `ObjectTransaction`); do not reintroduce a distributed lock. +- Minimal blast radius: the fix lands at the two chokepoint helpers. + +## Non-goals + +- Changing the one-entry-per-bucket storage model. +- Multi-filer-concurrent bucket writes (addressed only as an optional phase 3). + +## The two ingredients map to two complementary fixes + +### Fix A — serialize + read fresh (closes the window for whole-entry writers) + +Both `updateBucketConfig` and `UpdateBucketMetadata` must run their RMW under one +per-bucket critical section, and **re-read the entry fresh from the filer inside +it** — not rebuild from the cached `BucketConfig`. The lock alone is insufficient: +without the fresh read, two serialized writers still each apply a stale snapshot. + +### Fix B — field-level updates (removes the collision entirely) + +The two writers touch disjoint fields (`Extended[versioning]` vs `Content`). If +each path updated only its own field instead of rewriting the whole entry, neither +could clobber the other regardless of ordering. This is the structural fix and +makes serialization a defense-in-depth concern rather than a correctness +requirement for cross-field cases. + +## Where to serialize (layering) + +The bucket entry is a single filer entry, so unlike object writes there is no +sharding — the question is purely the scope of the lock: + +| Layer | Serializes across | Cost | Notes | +|---|---|---|---| +| 1. Gateway-local per-bucket lock | one gateway process | tiny | fixes the reported (single-gateway/CI) case | +| 2. Filer per-path lock via conditional write | all gateways on one filer | small | reuses #9640 `CreateEntry`+`WriteCondition` | +| 3. Route-by-key to bucket-key owner filer | all gateways and filers | medium | same mechanism as the object DLM-removal | + +## Recommended plan (phased) + +### Phase 1 — minimal fix for #9651 (gateway-local lock + fresh read) + +Add a bounded per-bucket lock table to `S3ApiServer`, reusing the same +`util.LockTable` the filer uses for its per-path lock: + +```go +// in S3ApiServer +bucketConfigLocks *util.LockTable[string] // serialize bucket-entry RMW + +func (s3a *S3ApiServer) withBucketConfigLock(bucket string, fn func() s3err.ErrorCode) s3err.ErrorCode { + lk := s3a.bucketConfigLocks.AcquireLock("bucketConfig", bucket, util.ExclusiveLock) + defer s3a.bucketConfigLocks.ReleaseLock(bucket, lk) + return fn() +} +``` + +Wrap the RMW in **both** chokepoints, and inside the lock read the entry fresh: + +- `updateBucketConfig`: acquire the lock; re-read `/buckets/` from the filer + (not the cache); rebuild `BucketConfig` from that fresh entry; apply `fn`; write; + invalidate cache; release. +- `UpdateBucketMetadata`/`setBucketMetadata`: same lock key; it already reads fresh, + so it just needs to share the critical section. + +Both must use the **same** lock keyed on `bucket`, so versioning and encryption +contend on one mutex. This closes the reported window. Limitation: only one +gateway; two gateways behind a load balancer still race. + +Test: parallel `PutBucketVersioning` + `PutBucketEncryption`, assert both persist +(the exact Terraform scenario), plus an N-way parallel variant over distinct +fields. + +### Phase 2 — robust across gateways (field-level + CAS via merged primitives) + +Move the writers off whole-entry rewrites: + +- **Extended-based config** (versioning, object-lock, ownership, tagging-in-Extended) + → `ObjectTransaction` `PATCH_EXTENDED` on `/buckets/`. The owner filer reads + the entry fresh under its per-path lock and merges only the named keys, so the + gateway never sends a whole-entry snapshot — this dissolves *both* ingredients for + these fields. +- **`Content`-based config** (encryption, CORS, tags blob) — **chosen and + implemented (b3): extend `PATCH_EXTENDED` with `set_content`.** Under the same + per-path lock the filer reads the entry fresh, merges extended attributes, and + replaces `Content`, preserving the rest. So a content write becomes a field-level + patch too — `setBucketMetadata` patches `Content`, `updateBucketConfig` patches + extended keys, and the two serialize on the lock instead of racing whole-entry + rewrites. This is cleaner than the alternatives below: no client-side retry, no + storage migration, and it reuses `ObjectTransaction`'s existing atomic lock. + - (b1, rejected) Conditional `CreateEntry` overwrite with `IF_ETAG_MATCH` + retry + (#9640): correct but needs client-side retry, and the bucket directory entry has + no reliable ETag to compare on. + - (b2, future) Migrate each per-feature config out of the single `Content` blob + into its own `Extended` key. Then even *intra-blob* writes (tags vs encryption) + stop racing. Larger migration; tracked separately. + +Once all paths are field-level patches, the phase-1 gateway lock is unnecessary — +the filer enforces atomicity. (This is the path taken: phase 1 was skipped.) + +### Phase 3 — multi-filer (only if needed) + +If multiple filers can write `/buckets/` concurrently, a filer-local per-path +lock no longer suffices. Route bucket-config writes to +`PrimaryForKey("/buckets/")` (the lock-ring view) and serialize on that one +owner filer — the same route-by-key design used to take object writes off the DLM. +Overkill for rare config writes; include only if multi-filer bucket writes are real. + +## Correctness summary + +- Phase 1: all RMW for a bucket serialize within a gateway; the fresh read means the + second writer observes the first's change. Closes #9651 for single-gateway. +- Phase 2: `PATCH_EXTENDED` is atomic field-level merge at the filer (no snapshot); + CAS turns a concurrent `Content` write into a retry, enforced under the filer's + per-path lock — correct for any number of gateways sharing a filer. +- Phase 3: one owner filer serializes all writers — correct across filers too. + +## Scope checklist (every path that RMWs the bucket entry) + +All of these funnel through the two chokepoints, so fixing the chokepoints covers +them — but the fix must not leave any of them on an unserialized path: + +- via `updateBucketConfig`: versioning, object-lock config, lifecycle, ACL/owner. +- via `UpdateBucketMetadata`/`setBucketMetadata`: encryption, CORS, tagging, + ownership controls, bucket policy, notification. +- bucket create/delete (`CreateEntry`/`DeleteEntry` of `/buckets/`) already + go through the filer's per-path lock on `CreateEntry`; ensure they take the same + bucket lock if they also patch config. + +## Cache rule (must document in code) + +Under the lock, **read the entry from the filer, never rebuild from the cached +`BucketConfig`**. The cache is for reads; it must be invalidated on every write and +never be the source for an RMW. This is the single most important detail — the lock +without the fresh read does not fix the bug. diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 10505f695..988e5f06b 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -312,6 +312,8 @@ message ObjectMutation { bool is_delete_data = 7; // DELETE: also delete chunk data bool is_recursive = 8; // DELETE: recurse into a directory Recompute recompute = 9; // RECOMPUTE_LATEST parameters + bool set_content = 10; // PATCH_EXTENDED: replace Entry.content with content + bytes content = 11; // PATCH_EXTENDED: new Entry.content when set_content } // Recompute re-derives a pointer entry (directory/name on the mutation) from the diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 10505f695..988e5f06b 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -312,6 +312,8 @@ message ObjectMutation { bool is_delete_data = 7; // DELETE: also delete chunk data bool is_recursive = 8; // DELETE: recurse into a directory Recompute recompute = 9; // RECOMPUTE_LATEST parameters + bool set_content = 10; // PATCH_EXTENDED: replace Entry.content with content + bytes content = 11; // PATCH_EXTENDED: new Entry.content when set_content } // Recompute re-derives a pointer entry (directory/name on the mutation) from the diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 17e630401..f0b2bbfd8 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1428,6 +1428,8 @@ type ObjectMutation struct { IsDeleteData bool `protobuf:"varint,7,opt,name=is_delete_data,json=isDeleteData,proto3" json:"is_delete_data,omitempty"` // DELETE: also delete chunk data IsRecursive bool `protobuf:"varint,8,opt,name=is_recursive,json=isRecursive,proto3" json:"is_recursive,omitempty"` // DELETE: recurse into a directory Recompute *Recompute `protobuf:"bytes,9,opt,name=recompute,proto3" json:"recompute,omitempty"` // RECOMPUTE_LATEST parameters + SetContent bool `protobuf:"varint,10,opt,name=set_content,json=setContent,proto3" json:"set_content,omitempty"` // PATCH_EXTENDED: replace Entry.content with content + Content []byte `protobuf:"bytes,11,opt,name=content,proto3" json:"content,omitempty"` // PATCH_EXTENDED: new Entry.content when set_content unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1525,6 +1527,20 @@ func (x *ObjectMutation) GetRecompute() *Recompute { return nil } +func (x *ObjectMutation) GetSetContent() bool { + if x != nil { + return x.SetContent + } + return false +} + +func (x *ObjectMutation) GetContent() []byte { + if x != nil { + return x.Content + } + return nil +} + // Recompute re-derives a pointer entry (directory/name on the mutation) from the // current contents of a scanned directory, atomically under the transaction's // lock. It is mechanical: the filer picks the child that sorts first or last by @@ -6315,7 +6331,7 @@ const file_filer_proto_rawDesc = "" + "\x13IF_UNMODIFIED_SINCE\x10\x05\x12\x15\n" + "\x11IF_MODIFIED_SINCE\x10\x06\x12\x19\n" + "\x15IF_EXTENDED_NOT_EQUAL\x10\a\x12\x1c\n" + - "\x18IF_EXTENDED_TIME_ELAPSED\x10\b\"\x96\x04\n" + + "\x18IF_EXTENDED_TIME_ELAPSED\x10\b\"\xd1\x04\n" + "\x0eObjectMutation\x121\n" + "\x04type\x18\x01 \x01(\x0e2\x1d.filer_pb.ObjectMutation.TypeR\x04type\x12\x1c\n" + "\tdirectory\x18\x02 \x01(\tR\tdirectory\x12\x12\n" + @@ -6325,7 +6341,11 @@ const file_filer_proto_rawDesc = "" + "\x0fdelete_extended\x18\x06 \x03(\tR\x0edeleteExtended\x12$\n" + "\x0eis_delete_data\x18\a \x01(\bR\fisDeleteData\x12!\n" + "\fis_recursive\x18\b \x01(\bR\visRecursive\x121\n" + - "\trecompute\x18\t \x01(\v2\x13.filer_pb.RecomputeR\trecompute\x1a>\n" + + "\trecompute\x18\t \x01(\v2\x13.filer_pb.RecomputeR\trecompute\x12\x1f\n" + + "\vset_content\x18\n" + + " \x01(\bR\n" + + "setContent\x12\x18\n" + + "\acontent\x18\v \x01(\fR\acontent\x1a>\n" + "\x10SetExtendedEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\fR\x05value:\x028\x01\"E\n" + diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index ef67e2cb3..851b80437 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -363,6 +363,14 @@ func (fs *FilerServer) applyObjectMutation(ctx context.Context, m *filer_pb.Obje for _, k := range m.DeleteExtended { delete(newEntry.Extended, k) } + if m.SetContent { + newEntry.Content = m.Content + // Keep FileSize consistent with content for files; some stores and + // tools read the attribute directly. Directories carry no file size. + if !newEntry.IsDirectory() { + newEntry.FileSize = uint64(len(m.Content)) + } + } if err := fs.filer.UpdateEntry(ctx, oldEntry, newEntry); err != nil { return err } diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index 50f5cdf57..d47eebf48 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/weed/server/filer_grpc_server_object_txn_test.go @@ -111,6 +111,83 @@ func TestObjectTransactionPatchNotifies(t *testing.T) { } } +// PATCH_EXTENDED with set_content replaces Entry.Content while merging Extended +// and preserving the rest; without set_content, Content is left untouched. +func TestObjectTransactionPatchContent(t *testing.T) { + now := time.Unix(1700000000, 0) + fs, store := newTxnTestServer(map[string]*filer.Entry{ + "/buckets/b": { + Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0755 | (1 << 31)}, + Extended: map[string][]byte{"versioning": []byte("Enabled")}, + Content: []byte("old-content"), + }, + }) + + // set_content replaces Content and merges an Extended key, preserving the + // existing versioning key. + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/b", + Mutations: []*filer_pb.ObjectMutation{ + {Type: filer_pb.ObjectMutation_PATCH_EXTENDED, Directory: "/buckets", Name: "b", + SetContent: true, Content: []byte("encryption-blob"), + SetExtended: map[string][]byte{"cors": []byte("yes")}}, + }, + }) + if err != nil || resp.Error != "" { + t.Fatalf("patch set_content failed: err=%v resp=%q", err, resp.Error) + } + e := store.entries["/buckets/b"] + if string(e.Content) != "encryption-blob" { + t.Fatalf("content = %q, want encryption-blob", e.Content) + } + if string(e.Extended["versioning"]) != "Enabled" || string(e.Extended["cors"]) != "yes" { + t.Fatalf("extended not merged: %v", e.Extended) + } + + // A PATCH without set_content must not disturb Content. + resp, err = fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/b", + Mutations: []*filer_pb.ObjectMutation{ + {Type: filer_pb.ObjectMutation_PATCH_EXTENDED, Directory: "/buckets", Name: "b", + SetExtended: map[string][]byte{"versioning": []byte("Suspended")}}, + }, + }) + if err != nil || resp.Error != "" { + t.Fatalf("patch extended-only failed: err=%v resp=%q", err, resp.Error) + } + e = store.entries["/buckets/b"] + if string(e.Content) != "encryption-blob" { + t.Fatalf("content clobbered by extended-only patch: %q", e.Content) + } + if string(e.Extended["versioning"]) != "Suspended" { + t.Fatalf("versioning = %q, want Suspended", e.Extended["versioning"]) + } + if e.FileSize != 0 { + t.Fatalf("directory FileSize must stay 0, got %d", e.FileSize) + } + + // For a file, set_content syncs FileSize to the new content length, even when + // the content shrinks. + store.entries["/file"] = &filer.Entry{ + FullPath: "/file", + Attr: filer.Attr{Inode: 9, Mtime: now, Crtime: now, Mode: 0644, FileSize: 100}, + Content: []byte("xxxxxxxxxxxxxxx"), + } + resp, err = fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/file", + Mutations: []*filer_pb.ObjectMutation{ + {Type: filer_pb.ObjectMutation_PATCH_EXTENDED, Directory: "/", Name: "file", + SetContent: true, Content: []byte("short")}, + }, + }) + if err != nil || resp.Error != "" { + t.Fatalf("file patch failed: err=%v resp=%q", err, resp.Error) + } + if f := store.entries["/file"]; string(f.Content) != "short" || f.FileSize != uint64(len("short")) { + t.Fatalf("file content=%q FileSize=%d, want short/5", f.Content, f.FileSize) + } +} + // A failing precondition aborts before any mutation is applied. func TestObjectTransactionPreconditionAborts(t *testing.T) { now := time.Unix(1700000000, 0)