s3: add blob offload rollout scaffolding

adapter/admin_grpc.go

@@ -60,8 +60,9 @@ func (n NodeIdentity) toProto() *pb.NodeIdentity {
 // GetClusterOverview and GetRaftGroups; remaining RPCs return Unimplemented so
 // the generated client can still compile against older nodes during rollout.
 type AdminServer struct {
-	self    NodeIdentity
-	members []NodeIdentity
+	self         NodeIdentity
+	members      []NodeIdentity
+	capabilities map[string]bool
 
 	groupsMu sync.RWMutex
 	groups   map[uint64]AdminGroup
@@ -89,10 +90,11 @@ type AdminServer struct {
 func NewAdminServer(self NodeIdentity, members []NodeIdentity) *AdminServer {
 	cloned := append([]NodeIdentity(nil), members...)
 	return &AdminServer{
-		self:    self,
-		members: cloned,
-		groups:  make(map[uint64]AdminGroup),
-		now:     time.Now,
+		self:         self,
+		members:      cloned,
+		capabilities: make(map[string]bool),
+		groups:       make(map[uint64]AdminGroup),
+		now:          time.Now,
 	}
 }
 
@@ -129,6 +131,20 @@ func (s *AdminServer) RegisterSampler(sampler KeyVizSampler) {
 	s.groupsMu.Unlock()
 }
 
+// SetCapability exposes a local binary or runtime capability in
+// GetClusterOverview. Disabled capabilities are kept in the map so fan-out
+// callers can distinguish "known false" from an older server that lacks the
+// field or key entirely.
+func (s *AdminServer) SetCapability(name string, enabled bool) {
+	name = strings.TrimSpace(name)
+	if s == nil || name == "" {
+		return
+	}
+	s.groupsMu.Lock()
+	s.capabilities[name] = enabled
+	s.groupsMu.Unlock()
+}
+
 // GetClusterOverview returns the local node identity, the current member
 // list, and per-group leader identity collected from the engines registered
 // via RegisterGroup. The member list is the union of (a) the bootstrap seed
@@ -145,9 +161,23 @@ func (s *AdminServer) GetClusterOverview(
 		Self:         s.self.toProto(),
 		Members:      members,
 		GroupLeaders: leaders,
+		Capabilities: s.snapshotCapabilities(),
 	}, nil
 }
 
+func (s *AdminServer) snapshotCapabilities() map[string]bool {
+	s.groupsMu.RLock()
+	defer s.groupsMu.RUnlock()
+	if len(s.capabilities) == 0 {
+		return nil
+	}
+	out := make(map[string]bool, len(s.capabilities))
+	for name, enabled := range s.capabilities {
+		out[name] = enabled
+	}
+	return out
+}
+
 // snapshotMembers unions the seed members with the live Configuration of each
 // registered group, preferring the live address when the same NodeID appears
 // in both sources. A stale bootstrap entry cannot outvote a readdressed node:

adapter/admin_grpc_test.go

@@ -71,6 +71,24 @@ func TestGetClusterOverviewReturnsSelfAndLeaders(t *testing.T) {
 	}
 }
 
+func TestGetClusterOverviewReturnsCapabilities(t *testing.T) {
+	t.Parallel()
+	srv := NewAdminServer(NodeIdentity{NodeID: "node-a"}, nil)
+	srv.SetCapability(S3BlobOffloadCapabilityName, false)
+	srv.SetCapability("feature-test", true)
+
+	resp, err := srv.GetClusterOverview(context.Background(), &pb.GetClusterOverviewRequest{})
+	if err != nil {
+		t.Fatalf("GetClusterOverview: %v", err)
+	}
+	if got, ok := resp.Capabilities[S3BlobOffloadCapabilityName]; !ok || got {
+		t.Fatalf("s3 blob offload capability = (%v, %v), want (false, true)", got, ok)
+	}
+	if got, ok := resp.Capabilities["feature-test"]; !ok || !got {
+		t.Fatalf("feature-test capability = (%v, %v), want (true, true)", got, ok)
+	}
+}
+
 func TestGetRaftGroupsExposesCommitApplied(t *testing.T) {
 	t.Parallel()
 	srv := NewAdminServer(NodeIdentity{NodeID: "n1"}, nil)

adapter/s3.go

@@ -111,6 +111,9 @@ type S3Server struct {
 	cleanupSem           chan struct{}
 	putAdmission         *s3PutAdmission
 	putAdmissionObserver S3PutAdmissionObserver
+	blobOffloadEnabled   bool
+	blobOffloadChecker   S3BlobOffloadCapabilityChecker
+	blobOffloadObserver  S3BlobOffloadObserver
 }
 
 type s3BucketMeta struct {
@@ -336,14 +339,15 @@ type s3ListPartEntry struct {
 
 func NewS3Server(listen net.Listener, s3Addr string, st store.MVCCStore, coordinate kv.Coordinator, leaderS3 map[string]string, opts ...S3ServerOption) *S3Server {
 	s := &S3Server{
-		listen:       listen,
-		s3Addr:       s3Addr,
-		region:       s3DefaultRegion,
-		store:        st,
-		coordinator:  kv.WithKeyVizLabel(coordinate, keyviz.LabelS3),
-		leaderS3:     cloneLeaderAddrMap(leaderS3),
-		cleanupSem:   make(chan struct{}, s3ManifestCleanupWorkers),
-		putAdmission: newS3PutAdmissionFromEnv(),
+		listen:             listen,
+		s3Addr:             s3Addr,
+		region:             s3DefaultRegion,
+		store:              st,
+		coordinator:        kv.WithKeyVizLabel(coordinate, keyviz.LabelS3),
+		leaderS3:           cloneLeaderAddrMap(leaderS3),
+		cleanupSem:         make(chan struct{}, s3ManifestCleanupWorkers),
+		putAdmission:       newS3PutAdmissionFromEnv(),
+		blobOffloadEnabled: newS3BlobOffloadEnabledFromEnv(),
 	}
 	for _, opt := range opts {
 		if opt != nil {
@@ -877,6 +881,7 @@ func (s *S3Server) putObject(w http.ResponseWriter, r *http.Request, bucket stri
 	if !s.admitS3PutRequest(w, r, bucket, objectKey, s3MaxObjectSizeBytes, "object exceeds maximum allowed size") {
 		return
 	}
+	s.observeS3BlobOffloadDecision(r.Context())
 	upload, uploadBodyErr, uploadErr := s.uploadS3ObjectData(
 		r.Context(), r, streamBody, state, bucket, objectKey, expectedPayloadSHA,
 	)
@@ -1228,6 +1233,7 @@ func (s *S3Server) uploadPart(w http.ResponseWriter, r *http.Request, bucket str
 	if !s.admitS3PutRequest(w, r, bucket, objectKey, s3MaxPartSizeBytes, "part exceeds maximum allowed size") {
 		return
 	}
+	s.observeS3BlobOffloadDecision(r.Context())
 	upload, previous, uploadBodyErr, uploadErr := s.storeS3UploadPart(
 		r.Context(), r, streamBody, state, bucket, objectKey, uploadID, admissionProtocol,
 	)

adapter/s3_admin.go

@@ -496,6 +496,7 @@ func bucketDeleteSafetyNetElems(bucket string, generation uint64) []*kv.Elem[kv.
 		{Op: kv.DelPrefix, Key: s3keys.UploadMetaPrefixForBucket(bucket, generation)},
 		{Op: kv.DelPrefix, Key: s3keys.UploadPartPrefixForBucket(bucket, generation)},
 		{Op: kv.DelPrefix, Key: s3keys.BlobPrefixForBucket(bucket, generation)},
+		{Op: kv.DelPrefix, Key: s3keys.ChunkRefPrefixForBucket(bucket, generation)},
 		{Op: kv.DelPrefix, Key: s3keys.GCUploadPrefixForBucket(bucket, generation)},
 		{Op: kv.DelPrefix, Key: s3keys.RoutePrefixForBucket(bucket, generation)},
 	}

adapter/s3_admin_test.go

@@ -386,7 +386,7 @@ func TestS3Server_AdminDeleteBucket_SweepsOrphansAcrossAllPerBucketPrefixes(t *t
 	gen := summary.Generation
 	require.NotZero(t, gen)
 
-	// Plant orphan keys across the 5 non-manifest per-bucket
+	// Plant orphan keys across the 6 non-manifest per-bucket
 	// prefixes. Each entry is what a concurrent PutObject (or its
 	// in-flight multipart upload state) would leave behind if it
 	// committed in the AdminDeleteBucket race window. The values
@@ -401,6 +401,7 @@ func TestS3Server_AdminDeleteBucket_SweepsOrphansAcrossAllPerBucketPrefixes(t *t
 		{Op: kv.Put, Key: s3keys.UploadMetaKey(bucket, gen, objectName, uploadID), Value: []byte("orphan-upload-meta")},
 		{Op: kv.Put, Key: s3keys.UploadPartKey(bucket, gen, objectName, uploadID, 1), Value: []byte("orphan-part")},
 		{Op: kv.Put, Key: s3keys.BlobKey(bucket, gen, objectName, uploadID, 1, 0), Value: []byte("orphan-chunk")},
+		{Op: kv.Put, Key: s3keys.ChunkRefKey(bucket, gen, objectName, uploadID, 1, 0), Value: []byte("orphan-chunk-ref")},
 		{Op: kv.Put, Key: s3keys.GCUploadKey(bucket, gen, objectName, uploadID), Value: []byte("orphan-gc")},
 		{Op: kv.Put, Key: s3keys.RouteKey(bucket, gen, objectName), Value: []byte("orphan-route")},
 	}
@@ -432,6 +433,7 @@ func TestS3Server_AdminDeleteBucket_SweepsOrphansAcrossAllPerBucketPrefixes(t *t
 		{"upload_meta", s3keys.UploadMetaPrefixForBucket(bucket, gen)},
 		{"upload_part", s3keys.UploadPartPrefixForBucket(bucket, gen)},
 		{"blob", s3keys.BlobPrefixForBucket(bucket, gen)},
+		{"chunk_ref", s3keys.ChunkRefPrefixForBucket(bucket, gen)},
 		{"gc_upload", s3keys.GCUploadPrefixForBucket(bucket, gen)},
 		{"route", s3keys.RoutePrefixForBucket(bucket, gen)},
 	}

adapter/s3_blob_offload.go

@@ -0,0 +1,118 @@
+package adapter
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"strconv"
+	"strings"
+)
+
+const (
+	// S3BlobOffloadCapabilityName is the Admin GetClusterOverview capability
+	// key used by mixed-version PUT admission before writing chunkref metadata.
+	S3BlobOffloadCapabilityName = "feature_s3_blob_offload"
+
+	s3BlobOffloadEnvVar = "ELASTICKV_S3_BLOB_OFFLOAD"
+
+	s3BlobOffloadModeLegacy  = "legacy"
+	s3BlobOffloadModeOffload = "offload"
+
+	s3BlobOffloadReasonFlagDisabled      = "flag_disabled"
+	s3BlobOffloadReasonCapabilityMissing = "capability_missing"
+	s3BlobOffloadReasonDataPathDisabled  = "data_path_disabled"
+	s3BlobOffloadReasonEnabled           = "enabled"
+)
+
+// S3BlobOffloadCapabilityChecker is the cluster-wide mixed-version guard for
+// admitting an S3 PUT into the offloaded chunkref/chunkblob keyspace.
+type S3BlobOffloadCapabilityChecker interface {
+	AllPeersSupportS3BlobOffload(ctx context.Context) bool
+}
+
+// S3BlobOffloadObserver records offload rollout and chunkblob durability
+// outcomes. The data path is still fail-closed, so only decision counters are
+// emitted today; the remaining counters are wired for the durable replication
+// PR that will turn the offload path on.
+type S3BlobOffloadObserver interface {
+	ObserveS3BlobOffloadDecision(mode, reason string)
+	ObserveS3ChunkBlobReplicationDegraded()
+	ObserveS3ChunkBlobSHAMismatch()
+	ObserveS3ChunkBlobUnrecoverable()
+}
+
+type s3BlobOffloadDecision struct {
+	mode   string
+	reason string
+}
+
+// S3BlobOffloadLocalCapability reports whether this binary can safely serve the
+// full offloaded S3 chunkref/chunkblob data path. The scaffolding PR keeps this
+// false so future leaders do not mistake these nodes for offload readers.
+func S3BlobOffloadLocalCapability() bool {
+	return false
+}
+
+func WithS3BlobOffloadEnabled(enabled bool) S3ServerOption {
+	return func(server *S3Server) {
+		if server == nil {
+			return
+		}
+		server.blobOffloadEnabled = enabled
+	}
+}
+
+func WithS3BlobOffloadCapabilityChecker(checker S3BlobOffloadCapabilityChecker) S3ServerOption {
+	return func(server *S3Server) {
+		if server == nil {
+			return
+		}
+		server.blobOffloadChecker = checker
+	}
+}
+
+func WithS3BlobOffloadObserver(observer S3BlobOffloadObserver) S3ServerOption {
+	return func(server *S3Server) {
+		if server == nil {
+			return
+		}
+		server.blobOffloadObserver = observer
+	}
+}
+
+func newS3BlobOffloadEnabledFromEnv() bool {
+	raw, ok := os.LookupEnv(s3BlobOffloadEnvVar)
+	if !ok {
+		return false
+	}
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return false
+	}
+	enabled, err := strconv.ParseBool(raw)
+	if err != nil {
+		slog.Warn("invalid S3 blob offload boolean env; using default", "name", s3BlobOffloadEnvVar, "value", raw, "default", false)
+		return false
+	}
+	return enabled
+}
+
+func (s *S3Server) s3BlobOffloadDecision(ctx context.Context) s3BlobOffloadDecision {
+	if s == nil || !s.blobOffloadEnabled {
+		return s3BlobOffloadDecision{mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonFlagDisabled}
+	}
+	if s.blobOffloadChecker == nil || !s.blobOffloadChecker.AllPeersSupportS3BlobOffload(ctx) {
+		return s3BlobOffloadDecision{mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonCapabilityMissing}
+	}
+	if !S3BlobOffloadLocalCapability() {
+		return s3BlobOffloadDecision{mode: s3BlobOffloadModeLegacy, reason: s3BlobOffloadReasonDataPathDisabled}
+	}
+	return s3BlobOffloadDecision{mode: s3BlobOffloadModeOffload, reason: s3BlobOffloadReasonEnabled}
+}
+
+func (s *S3Server) observeS3BlobOffloadDecision(ctx context.Context) {
+	decision := s.s3BlobOffloadDecision(ctx)
+	if s != nil && s.blobOffloadObserver != nil {
+		s.blobOffloadObserver.ObserveS3BlobOffloadDecision(decision.mode, decision.reason)
+	}
+}