feat(s3): implement object lock (per-key version-state tree)

Implements docs/s3-object-lock.md. WORM enforcement lives in versitygw's
controller (auth.CheckObjectAccess); the backend stores lock state and
returns the exact absent-case sentinels.

Per-version retention and legal holds live in a per-key version-state
tree beside the prev tree: ObjectLeaf.State roots an MST mapping
revSeqKey(seq) to VersionState blocks under the new "/versionstate/0"
union arm. The State field joins ObjectLeaf under the existing
"/objectleaf/0" key with no compatibility shim, per the repo's dev-only
data posture. Mutations rewrite only positional blocks, so manifests
stay immutable and versioning invariant 5 holds verbatim. VersionState
reserves the Tags field for object tagging (handlers, not format, in the
follow-up). Creation-time stamping (PutObject / CopyObject /
CompleteMultipartUpload lock headers) and delete-path cleanup run inside
the same commits that create and remove versions. The bucket-level lock
configuration is a registry column beside versioning; CreateBucket with
x-amz-bucket-object-lock-enabled creates the bucket versioned and locked
in one insert, and PutBucketVersioning refuses to suspend a lock bucket.

Conformance notes pinned by the run: the creation-time header paths
report the NoSpaces variant of the missing-configuration error while the
four per-version methods report the spaced one; key existence outranks
the lock-enabled gate (GetObjectRetention_non_existing_object); the
controller passes an absent retain-until header as a pointer to the zero
time; and NoSuchVersionError / InvalidArgumentError now pass through
mapCommitError verbatim like PreconditionFailedError. itest gains the six
lock groups, a WORMProtection xfail table (bucket-policy-dependent bypass
cases), and a LockCreation category for lock-enabled-bucket cases from
plain-conf groups that need the versioned teardown; the versioning tables
absorb the Versioning_* lock rows and promote
DeleteObject_non_existing_objects.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: frrist

bucket/cbor_gen.go

@@ -19,8 +19,9 @@ type VersionNode struct {
 
 // ObjectLeaf is the per-key version group, stored under the "/objectleaf/0"
 // union key (docs/s3-versioning.md §2.1). A key's top-MST value points at one
-// once the key has been superseded; until then the value is its manifest
-// (under "/objectmanifest/0").
+// once the key has been superseded or has received version state
+// (docs/s3-object-lock.md §4.1); until then the value is its manifest (under
+// "/objectmanifest/0").
 type ObjectLeaf struct {
 	// Current is the head version — what GET/HEAD/ListObjects resolve with
 	// a single leaf read, no descent into Prev.
@@ -36,6 +37,46 @@ type ObjectLeaf struct {
 	// because Prev is keyed by Seq and the null version's id ("null") does
 	// not encode its Seq.
 	NullSeq uint64 `cborgen:"n"`
+
+	// State is the root of the per-key version-state MST: revSeqKey(seq) →
+	// the version's VersionState block CID (docs/s3-object-lock.md §4.1).
+	// Nil when no version of this key carries explicit state, which is what
+	// makes state-free keys free to read.
+	State *cid.Cid `cborgen:"st"`
+}
+
+// LegalHold values for VersionState.LegalHold. S3 distinguishes never-set
+// (a 400 from GetObjectLegalHold) from an explicit OFF (a 200), so the
+// stored state is tri-valued.
+const (
+	LegalHoldUnset uint8 = 0
+	LegalHoldOff   uint8 = 1
+	LegalHoldOn    uint8 = 2
+)
+
+// VersionState is one version's mutable service state, stored as its own
+// catalog block under the "/versionstate/0" union key (docs/s3-object-lock.md
+// §4.1). Object lock is its first tenant; the block is the home for
+// per-version state that mutates after the version is created. Mutations
+// merge: each operation replaces the fields it owns and carries every other
+// field verbatim; a mutation that leaves every field absent deletes the
+// version's state-tree entry.
+type VersionState struct {
+	// Retention is the stored retention document (docs/s3-object-lock.md §2),
+	// verbatim; nil when retention has never been set.
+	Retention []byte `cborgen:"r"`
+	// LegalHold is tri-valued: LegalHoldUnset, LegalHoldOff, LegalHoldOn.
+	LegalHold uint8 `cborgen:"h"`
+	// Tags is reserved for object tagging (planned follow-up). Declared now
+	// so every rewrite round-trips it under the merge rule; the tagging
+	// feature adds handlers, not format. Nil until it lands.
+	Tags map[string]string `cborgen:"t"`
+}
+
+// Empty reports whether every field is absent — the elision condition: an
+// empty state block is removed from the tree rather than stored.
+func (s *VersionState) Empty() bool {
+	return s.Retention == nil && s.LegalHold == LegalHoldUnset && len(s.Tags) == 0
 }
 
 // ValueUnion is the keyed union every catalog value block is encoded as
@@ -144,3 +185,38 @@ func (e *EnvelopedLeaf) UnmarshalCBOR(r io.Reader) error {
 	e.Leaf = v.Leaf
 	return nil
 }
+
+// StateUnion is the keyed union for version-state blocks — the values of a
+// leaf's state tree (docs/s3-object-lock.md §4.1). A separate union from
+// ValueUnion so a state block decoded through ObjectValue (or vice versa)
+// fails loudly with no arm matched. cbor-gen generates its codec; decode
+// through EnvelopedVersionState, which requires the arm.
+type StateUnion struct {
+	State *VersionState `cborgen:"/versionstate/0,omitempty"`
+}
+
+// EnvelopedVersionState reads/writes one version-state block under its
+// "/versionstate/0" union key.
+type EnvelopedVersionState struct {
+	State *VersionState
+}
+
+func (e *EnvelopedVersionState) MarshalCBOR(w io.Writer) error {
+	if e.State == nil {
+		return fmt.Errorf("bucket: version-state block must carry a state")
+	}
+	u := StateUnion{State: e.State}
+	return u.MarshalCBOR(w)
+}
+
+func (e *EnvelopedVersionState) UnmarshalCBOR(r io.Reader) error {
+	var u StateUnion
+	if err := u.UnmarshalCBOR(r); err != nil {
+		return err
+	}
+	if u.State == nil {
+		return fmt.Errorf("bucket: block is not an enveloped version state")
+	}
+	e.State = u.State
+	return nil
+}

bucket/leaf.go

@@ -24,10 +24,12 @@ func testCid(t *testing.T, s string) cid.Cid {
 // decodes back with its fields intact.
 func TestObjectValue_LeafRoundTrip(t *testing.T) {
 	prev := testCid(t, "prev-root")
+	state := testCid(t, "state-root")
 	leaf := &ObjectLeaf{
 		Current: VersionNode{Seq: 9, VersionID: "01ARZ3NDEKTSV4RRFFQ69G5FAV", Manifest: testCid(t, "mf")},
 		Prev:    &prev,
 		NullSeq: 4,
+		State:   &state,
 	}
 	var buf bytes.Buffer
 	if err := LeafValue(leaf).MarshalCBOR(&buf); err != nil {
@@ -46,7 +48,8 @@ func TestObjectValue_LeafRoundTrip(t *testing.T) {
 	}
 	got := val.Leaf
 	if got.Current != leaf.Current || got.NullSeq != leaf.NullSeq ||
-		got.Prev == nil || !got.Prev.Equals(prev) {
+		got.Prev == nil || !got.Prev.Equals(prev) ||
+		got.State == nil || !got.State.Equals(state) {
 		t.Fatalf("leaf round-trip = %+v, want %+v", got, leaf)
 	}
 
@@ -148,3 +151,53 @@ func TestObjectValue_RejectsNonUnionBlocks(t *testing.T) {
 		t.Fatal("marshal of zero-arm value: err = nil, want rejection")
 	}
 }
+
+
+// TestVersionState_RoundTrip pins the version-state block: the
+// "/versionstate/0" union key, the tri-valued hold, the verbatim retention
+// bytes, and the reserved Tags field surviving a decode/re-encode cycle (the
+// §4.1 merge rule depends on unowned fields being carried).
+func TestVersionState_RoundTrip(t *testing.T) {
+	st := &VersionState{
+		Retention: []byte(`{"Mode":"GOVERNANCE","RetainUntilDate":"2027-01-02T15:04:05Z"}`),
+		LegalHold: LegalHoldOff,
+		Tags:      map[string]string{"team": "forge", "env": "dev"},
+	}
+	var buf bytes.Buffer
+	if err := (&EnvelopedVersionState{State: st}).MarshalCBOR(&buf); err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	if buf.Bytes()[0] != 0xa1 {
+		t.Fatalf("header = %#x, want 0xa1 (single-entry map)", buf.Bytes()[0])
+	}
+
+	var env EnvelopedVersionState
+	if err := env.UnmarshalCBOR(bytes.NewReader(buf.Bytes())); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	got := env.State
+	if !bytes.Equal(got.Retention, st.Retention) || got.LegalHold != LegalHoldOff ||
+		got.Tags["team"] != "forge" || got.Tags["env"] != "dev" || len(got.Tags) != 2 {
+		t.Fatalf("round-trip = %+v, want %+v", got, st)
+	}
+	if got.Empty() {
+		t.Fatal("populated state reports Empty")
+	}
+	if !(&VersionState{}).Empty() {
+		t.Fatal("zero state does not report Empty")
+	}
+
+	// The strict decoders reject cross-type blocks: a value block is not a
+	// state block, and a state block is not a value block.
+	var lbuf bytes.Buffer
+	if err := LeafValue(&ObjectLeaf{Current: VersionNode{Manifest: testCid(t, "m")}}).MarshalCBOR(&lbuf); err != nil {
+		t.Fatalf("leaf marshal: %v", err)
+	}
+	if err := env.UnmarshalCBOR(bytes.NewReader(lbuf.Bytes())); err == nil {
+		t.Fatal("EnvelopedVersionState.UnmarshalCBOR(leaf block) = nil error, want rejection")
+	}
+	var val ObjectValue
+	if err := val.UnmarshalCBOR(bytes.NewReader(buf.Bytes())); err == nil {
+		t.Fatal("ObjectValue.UnmarshalCBOR(state block) = nil error, want rejection")
+	}
+}

bucket/leaf_test.go

@@ -20,6 +20,8 @@ func main() {
 		bucket.ObjectLeaf{},
 		bucket.VersionNode{},
 		bucket.ValueUnion{},
+		bucket.VersionState{},
+		bucket.StateUnion{},
 	); err != nil {
 		panic(err)
 	}

gen/main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/multiformats/go-multihash"
 
 	"github.com/fil-forge/ingot/blockstore"
+	"github.com/fil-forge/ingot/registry"
 	"github.com/fil-forge/libforge/testutil"
 )
 
@@ -28,7 +29,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
 	ctx := context.Background()
 	m := NewMemStore()
 
-	if err := m.Create(ctx, "bk", testutil.RandomDID(t)); err != nil {
+	if err := m.Create(ctx, "bk", testutil.RandomDID(t), registry.CreateState{}); err != nil {
 		t.Fatalf("Create: %v", err)
 	}
 	committed := testCid(t, "committed-root")
@@ -61,7 +62,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
 
 	// A segment carrying ONLY a stale op-root must not advance forge_root at all.
 	m2 := NewMemStore()
-	_ = m2.Create(ctx, "bk2", testutil.RandomDID(t))
+	_ = m2.Create(ctx, "bk2", testutil.RandomDID(t), registry.CreateState{})
 	_ = m2.CASRoot(ctx, "bk2", cid.Undef, committed)
 	seq2, _ := m2.NextSegmentSeq(ctx)
 	_ = m2.InsertSegmentOpen(ctx, blockstore.PlaneCatalog, seq2, "bk2")

inmem/segments_test.go

@@ -163,18 +163,23 @@ func (m *MemStore) ListBuckets(ctx context.Context, req s3.Request) (*bucket.Lis
 
 // Registry methods ===========================================================
 
-func (m *MemStore) Create(_ context.Context, name string, space did.DID) error {
+func (m *MemStore) Create(_ context.Context, name string, space did.DID, init registry.CreateState) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if _, ok := m.buckets[name]; ok {
 		return registry.ErrExists
 	}
+	v := init.Versioning
+	if v == "" {
+		v = registry.VersioningUnversioned
+	}
 	// Stamped here for parity with the buckets.created_at column default.
 	m.buckets[name] = &registry.State{
-		Name:       name,
-		Space:      space,
-		Versioning: registry.VersioningUnversioned,
-		CreatedAt:  time.Now().UTC(),
+		Name:             name,
+		Space:            space,
+		Versioning:       v,
+		ObjectLockConfig: init.ObjectLockConfig,
+		CreatedAt:        time.Now().UTC(),
 	}
 	return nil
 }
@@ -240,6 +245,17 @@ func (m *MemStore) SetVersioning(_ context.Context, name string, v registry.Vers
 	return nil
 }
 
+func (m *MemStore) SetObjectLockConfig(_ context.Context, name string, cfg []byte) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	s, ok := m.buckets[name]
+	if !ok {
+		return registry.ErrNotFound
+	}
+	s.ObjectLockConfig = cfg
+	return nil
+}
+
 func (m *MemStore) AllocVersionSeq(_ context.Context, name string) (uint64, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

inmem/store.go

@@ -8,6 +8,8 @@ import (
 	s3 "github.com/fil-forge/libforge/commands/s3"
 	"github.com/fil-forge/libforge/commands/s3/bucket"
 	"github.com/fil-forge/libforge/testutil"
+
+	"github.com/fil-forge/ingot/registry"
 )
 
 // TestMemStoreListBucketsPagination covers the local pagination semantics of
@@ -20,7 +22,7 @@ func TestMemStoreListBucketsPagination(t *testing.T) {
 	m := NewMemStore()
 	for _, name := range []string{"apple", "apricot", "banana", "cherry"} {
 		space := testutil.RandomDID(t)
-		if err := m.Create(ctx, name, space); err != nil {
+		if err := m.Create(ctx, name, space, registry.CreateState{}); err != nil {
 			t.Fatalf("Create %q: %v", name, err)
 		}
 	}

inmem/store_test.go

@@ -12,6 +12,7 @@ import (
 // the in-memory harness was removed).
 
 var createBucketPass = []forgeCase{
+	{name: "default_object_lock", fn: integration.CreateBucket_default_object_lock},
 	{name: "invalid_bucket_name", fn: integration.CreateBucket_invalid_bucket_name},
 	{name: "invalid_canned_acl", fn: integration.CreateBucket_invalid_canned_acl},
 	{name: "invalid_ownership", fn: integration.CreateBucket_invalid_ownership},
@@ -27,7 +28,6 @@ var createBucketXFail = []forgeCase{
 	{name: "invalid_location_constraint", fn: integration.CreateBucket_invalid_location_constraint},
 	{name: "as_user", fn: integration.CreateBucket_as_user},
 	{name: "default_acl", fn: integration.CreateBucket_default_acl},
-	{name: "default_object_lock", fn: integration.CreateBucket_default_object_lock},
 	{name: "duplicate_keys", fn: integration.CreateBucket_duplicate_keys},
 	{name: "existing_bucket", fn: integration.CreateBucket_existing_bucket},
 	{name: "invalid_tags", fn: integration.CreateBucket_invalid_tags},