From eaaa08bb45fd0fb73f46015024b2457cdc3065ef Mon Sep 17 00:00:00 2001 From: bootjp Date: Sun, 19 Jul 2026 21:05:43 +0900 Subject: [PATCH] Add asynchronous S3 blob backfill --- adapter/s3.go | 11 + adapter/s3_blob_backfill.go | 592 +++++++++++++++++++++++++++++++ adapter/s3_blob_backfill_test.go | 352 ++++++++++++++++++ adapter/s3_blob_fetch.go | 84 ++++- adapter/s3_blob_offload.go | 9 + adapter/s3_blob_read.go | 12 +- kv/shard_store.go | 19 + kv/shard_store_test.go | 14 + main.go | 40 ++- main_encryption_write_wiring.go | 4 +- main_s3.go | 45 ++- main_s3_test.go | 56 +++ monitoring/s3.go | 40 +++ monitoring/s3_test.go | 12 + 14 files changed, 1262 insertions(+), 28 deletions(-) create mode 100644 adapter/s3_blob_backfill.go create mode 100644 adapter/s3_blob_backfill_test.go diff --git a/adapter/s3.go b/adapter/s3.go index 86099e517..f920980fb 100644 --- a/adapter/s3.go +++ b/adapter/s3.go @@ -119,6 +119,7 @@ type S3Server struct { blobLocalStores S3BlobLocalStoreResolver blobMinReplicas int blobPushBlocked func() bool + blobBackfiller *S3BlobBackfiller } type s3BucketMeta struct { @@ -402,6 +403,9 @@ func (s *S3Server) Run() error { } func (s *S3Server) Stop() { + if s != nil && s.blobBackfiller != nil { + s.blobBackfiller.Stop() + } if s != nil && s.blobCluster != nil { _ = s.blobCluster.Close() } @@ -410,6 +414,13 @@ func (s *S3Server) Stop() { } } +func (s *S3Server) StartBlobBackfill(ctx context.Context) error { + if s == nil || s.blobBackfiller == nil { + return nil + } + return s.blobBackfiller.Start(ctx, s) +} + func (s *S3Server) handle(w http.ResponseWriter, r *http.Request) { if serveS3Healthz(w, r) { return diff --git a/adapter/s3_blob_backfill.go b/adapter/s3_blob_backfill.go new file mode 100644 index 000000000..781d8eb65 --- /dev/null +++ b/adapter/s3_blob_backfill.go @@ -0,0 +1,592 @@ +package adapter + +import ( + "bytes" + "context" + "crypto/sha256" + "log/slog" + "math" + "os" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/kv" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const ( + s3BlobBackfillWorkersEnvVar = "ELASTICKV_S3_CHUNKBLOB_FETCH_WORKERS" + s3BlobBackfillQueueSizeEnvVar = "ELASTICKV_S3_CHUNKBLOB_FETCH_QUEUE_SIZE" + s3BlobBackfillRateEnvVar = "ELASTICKV_S3_CHUNKBLOB_FETCH_RATE_PER_PEER" + s3BlobBackfillBurstEnvVar = "ELASTICKV_S3_CHUNKBLOB_FETCH_BURST_PER_PEER" + s3BlobBackfillScanIntervalEnvVar = "ELASTICKV_S3_CHUNKBLOB_BACKFILL_SCAN_INTERVAL" + + s3BlobBackfillDefaultWorkers = 4 + s3BlobBackfillDefaultQueueSize = 4096 + s3BlobBackfillDefaultRate = 32 + s3BlobBackfillDefaultBurst = 8 + s3BlobBackfillDefaultScanInterval = time.Minute + s3BlobBackfillDefaultScanPageSize = 256 + s3BlobBackfillDefaultMaxAttempts = 3 + s3BlobBackfillRetryInitial = 100 * time.Millisecond + s3BlobBackfillRetryMax = time.Second + s3BlobBackfillRetryFactor = 2 + + s3BlobBackfillResultFetched = "fetched" + s3BlobBackfillResultLocalHit = "local_hit" + s3BlobBackfillResultGone = "gone" + s3BlobBackfillResultFailed = "failed" + s3BlobBackfillResultQueueDrop = "queue_drop" +) + +// S3BlobBackfillConfig bounds follower catch-up work independently from the +// Raft apply loop and from user-path proxy-on-miss reads. +type S3BlobBackfillConfig struct { + Workers int + QueueSize int + RatePerPeer int + BurstPerPeer int + ScanInterval time.Duration + ScanPageSize int + MaxAttempts int + RetryInitial time.Duration + RetryMax time.Duration +} + +// S3BlobBackfillConfigFromEnv loads the operational bounds for async follower +// blob fetch. Invalid or non-positive values fail startup instead of silently +// removing a resource bound. +func S3BlobBackfillConfigFromEnv() (S3BlobBackfillConfig, error) { + cfg := defaultS3BlobBackfillConfig() + var err error + if cfg.Workers, err = positiveEnvInt(s3BlobBackfillWorkersEnvVar, cfg.Workers); err != nil { + return S3BlobBackfillConfig{}, err + } + if cfg.QueueSize, err = positiveEnvInt(s3BlobBackfillQueueSizeEnvVar, cfg.QueueSize); err != nil { + return S3BlobBackfillConfig{}, err + } + if cfg.RatePerPeer, err = positiveEnvInt(s3BlobBackfillRateEnvVar, cfg.RatePerPeer); err != nil { + return S3BlobBackfillConfig{}, err + } + if cfg.BurstPerPeer, err = positiveEnvInt(s3BlobBackfillBurstEnvVar, cfg.BurstPerPeer); err != nil { + return S3BlobBackfillConfig{}, err + } + rawInterval := strings.TrimSpace(os.Getenv(s3BlobBackfillScanIntervalEnvVar)) + if rawInterval != "" { + cfg.ScanInterval, err = time.ParseDuration(rawInterval) + if err != nil || cfg.ScanInterval <= 0 { + return S3BlobBackfillConfig{}, errors.WithStack( + errors.Newf("%s must be a positive duration", s3BlobBackfillScanIntervalEnvVar), + ) + } + } + return cfg, nil +} + +func positiveEnvInt(name string, fallback int) (int, error) { + raw := strings.TrimSpace(os.Getenv(name)) + if raw == "" { + return fallback, nil + } + value, err := strconv.Atoi(raw) + if err != nil || value <= 0 { + return 0, errors.WithStack(errors.Newf("%s must be a positive integer", name)) + } + return value, nil +} + +func defaultS3BlobBackfillConfig() S3BlobBackfillConfig { + return S3BlobBackfillConfig{ + Workers: s3BlobBackfillDefaultWorkers, + QueueSize: s3BlobBackfillDefaultQueueSize, + RatePerPeer: s3BlobBackfillDefaultRate, + BurstPerPeer: s3BlobBackfillDefaultBurst, + ScanInterval: s3BlobBackfillDefaultScanInterval, + ScanPageSize: s3BlobBackfillDefaultScanPageSize, + MaxAttempts: s3BlobBackfillDefaultMaxAttempts, + RetryInitial: s3BlobBackfillRetryInitial, + RetryMax: s3BlobBackfillRetryMax, + } +} + +func normalizeS3BlobBackfillConfig(cfg S3BlobBackfillConfig) S3BlobBackfillConfig { + defaults := defaultS3BlobBackfillConfig() + if cfg.Workers <= 0 { + cfg.Workers = defaults.Workers + } + if cfg.QueueSize <= 0 { + cfg.QueueSize = defaults.QueueSize + } + if cfg.RatePerPeer <= 0 { + cfg.RatePerPeer = defaults.RatePerPeer + } + if cfg.BurstPerPeer <= 0 { + cfg.BurstPerPeer = defaults.BurstPerPeer + } + if cfg.ScanInterval <= 0 { + cfg.ScanInterval = defaults.ScanInterval + } + if cfg.ScanPageSize <= 0 { + cfg.ScanPageSize = defaults.ScanPageSize + } + if cfg.MaxAttempts <= 0 { + cfg.MaxAttempts = defaults.MaxAttempts + } + if cfg.RetryInitial <= 0 { + cfg.RetryInitial = defaults.RetryInitial + } + if cfg.RetryMax < cfg.RetryInitial { + cfg.RetryMax = max(defaults.RetryMax, cfg.RetryInitial) + } + return cfg +} + +// S3BlobLocalStoreScanner exposes every local shard store so startup and +// periodic scans can recover chunkrefs not observed through the live apply +// hook, including refs installed by snapshot restore. +type S3BlobLocalStoreScanner interface { + LocalStores() []store.MVCCStore +} + +// S3BlobBackfillObserver is an optional extension implemented by the metrics +// observer. Keeping it separate preserves compatibility with small test and +// embedding observers that only implement the M1 metric surface. +type S3BlobBackfillObserver interface { + ObserveS3ChunkBlobBackfillQueueDepth(depth int) + ObserveS3ChunkBlobBackfillResult(result string) +} + +// S3BlobBackfiller receives non-blocking FSM apply notifications and performs +// all reads, peer RPCs, verification, retries, and local writes on bounded +// background workers. +type S3BlobBackfiller struct { + config S3BlobBackfillConfig + queue chan []byte + pending sync.Map + + mu sync.Mutex + cancel context.CancelFunc + started bool + wg sync.WaitGroup + server atomic.Pointer[S3Server] + + limitersMu sync.Mutex + limiters map[string]*s3BlobTokenBucket +} + +var _ kv.ApplyObserver = (*S3BlobBackfiller)(nil) + +func NewS3BlobBackfiller(cfg S3BlobBackfillConfig) *S3BlobBackfiller { + cfg = normalizeS3BlobBackfillConfig(cfg) + return &S3BlobBackfiller{ + config: cfg, + queue: make(chan []byte, cfg.QueueSize), + limiters: make(map[string]*s3BlobTokenBucket), + } +} + +// OnApply stays non-blocking because it runs inline on the Raft apply +// goroutine. A full queue drops the notification; the periodic scanner is the +// durable recovery path for that case. +func (b *S3BlobBackfiller) OnApply(op pb.Op, key []byte) { + if b == nil || op != pb.Op_PUT { + return + } + if _, _, _, _, _, _, _, ok := s3keys.ParseVersionedChunkRefKey(key); !ok { + return + } + b.enqueue(context.Background(), key, false) +} + +func (b *S3BlobBackfiller) Start(ctx context.Context, server *S3Server) error { + if b == nil { + return nil + } + if server == nil || server.blobCluster == nil || server.blobLocalStores == nil { + return errors.New("s3 blob backfill data path is not configured") + } + if _, ok := server.blobLocalStores.(S3BlobLocalStoreScanner); !ok { + return errors.New("s3 blob backfill local store scanner is not configured") + } + b.mu.Lock() + defer b.mu.Unlock() + if b.started { + return nil + } + runCtx, cancel := context.WithCancel(ctx) + b.server.Store(server) + b.cancel = cancel + b.started = true + for range b.config.Workers { + b.wg.Add(1) + go b.runWorker(runCtx) + } + b.wg.Add(1) + go b.runScanner(runCtx) + return nil +} + +func (b *S3BlobBackfiller) Stop() { + if b == nil { + return + } + b.mu.Lock() + cancel := b.cancel + b.cancel = nil + b.started = false + b.mu.Unlock() + if cancel != nil { + cancel() + } + b.wg.Wait() + b.server.Store(nil) + b.clearPendingQueue() +} + +func (b *S3BlobBackfiller) clearPendingQueue() { + for { + select { + case key := <-b.queue: + b.pending.Delete(string(key)) + default: + b.pending.Range(func(key, _ any) bool { + b.pending.Delete(key) + return true + }) + return + } + } +} + +func (b *S3BlobBackfiller) enqueue(ctx context.Context, key []byte, wait bool) bool { + identity := string(key) + if _, loaded := b.pending.LoadOrStore(identity, struct{}{}); loaded { + return true + } + queued := bytes.Clone(key) + if wait { + select { + case b.queue <- queued: + b.observeQueueDepth() + return true + case <-ctx.Done(): + b.pending.Delete(identity) + return false + } + } + select { + case b.queue <- queued: + b.observeQueueDepth() + return true + default: + b.pending.Delete(identity) + b.observeResult(s3BlobBackfillResultQueueDrop) + return false + } +} + +func (b *S3BlobBackfiller) runWorker(ctx context.Context) { + defer b.wg.Done() + for { + select { + case <-ctx.Done(): + return + case key := <-b.queue: + b.observeQueueDepth() + b.backfillWithRetry(ctx, key) + b.pending.Delete(string(key)) + } + } +} + +func (b *S3BlobBackfiller) backfillWithRetry(ctx context.Context, key []byte) { + backoff := b.config.RetryInitial + var lastErr error + for attempt := 1; attempt <= b.config.MaxAttempts; attempt++ { + result, err := b.backfillOne(ctx, key) + if err == nil { + b.observeResult(result) + return + } + lastErr = err + if attempt == b.config.MaxAttempts || !waitS3BlobBackfillRetry(ctx, backoff) { + break + } + backoff = min(backoff*s3BlobBackfillRetryFactor, b.config.RetryMax) + } + if ctx.Err() == nil { + b.observeResult(s3BlobBackfillResultFailed) + slog.Warn("s3 chunkblob backfill failed", "key", string(key), "err", lastErr) + } +} + +func waitS3BlobBackfillRetry(ctx context.Context, delay time.Duration) bool { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +func (b *S3BlobBackfiller) backfillOne(ctx context.Context, refKey []byte) (string, error) { + server := b.server.Load() + if server == nil { + return "", errors.New("s3 blob backfill server is not started") + } + metadataStore, ok := server.blobLocalStores.LocalStoreForKey(refKey) + if !ok || metadataStore == nil { + return "", errors.New("resolve local s3 chunkref store") + } + ref, refCommitTS, gone, err := loadS3BlobBackfillRef(ctx, metadataStore, refKey) + if gone { + return s3BlobBackfillResultGone, nil + } + if err != nil { + return "", err + } + exists, err := server.localS3ChunkBlobExists(ctx, refKey, ref.ContentSHA256) + if err != nil { + return "", err + } + if exists { + return s3BlobBackfillResultLocalHit, nil + } + payload, err := b.fetchFromPeers(ctx, refKey, ref) + if err != nil { + return "", err + } + if err := server.storeFetchedS3ChunkBlob(ctx, refKey, ref.ContentSHA256, payload, refCommitTS); err != nil { + return "", err + } + return s3BlobBackfillResultFetched, nil +} + +func loadS3BlobBackfillRef( + ctx context.Context, + metadataStore store.MVCCStore, + refKey []byte, +) (s3keys.ChunkRefValue, uint64, bool, error) { + value, err := metadataStore.GetAt(ctx, refKey, math.MaxUint64) + if errors.Is(err, store.ErrKeyNotFound) { + return s3keys.ChunkRefValue{}, 0, true, nil + } + if err != nil { + return s3keys.ChunkRefValue{}, 0, false, errors.Wrap(err, "read local s3 chunkref") + } + ref, ok := s3keys.DecodeChunkRefValue(value) + if !ok { + return s3keys.ChunkRefValue{}, 0, false, errors.New("decode local s3 chunkref") + } + refCommitTS, exists, err := metadataStore.LatestCommitTS(ctx, refKey) + if err != nil { + return s3keys.ChunkRefValue{}, 0, false, errors.Wrap(err, "read local s3 chunkref timestamp") + } + if !exists || refCommitTS == 0 { + return s3keys.ChunkRefValue{}, 0, true, nil + } + return ref, refCommitTS, false, nil +} + +func (b *S3BlobBackfiller) fetchFromPeers(ctx context.Context, refKey []byte, ref s3keys.ChunkRefValue) ([]byte, error) { + server := b.server.Load() + if server == nil { + return nil, errors.New("s3 blob backfill server is not started") + } + current, discoveryErr := server.blobCluster.ReplicasForChunk(ctx, refKey) + writeTime := writeTimeS3BlobReplicas(ref.ReplicaPeers) + if discoveryErr != nil && len(writeTime) == 0 { + return nil, errors.Wrap(discoveryErr, "resolve s3 chunkblob backfill replicas") + } + peers := orderedS3BlobPeers(writeTime, current, server.blobCluster.SelfNodeID(), ref.SourcePeer) + for _, peer := range peers { + payload, ok, err := b.fetchFromPeer(ctx, server, peer, ref) + if err != nil { + return nil, err + } + if ok { + return payload, nil + } + } + return nil, errors.New("s3 chunkblob backfill failed on every replica") +} + +func (b *S3BlobBackfiller) fetchFromPeer( + ctx context.Context, + server *S3Server, + peer S3BlobReplica, + ref s3keys.ChunkRefValue, +) ([]byte, bool, error) { + if err := b.limiterFor(peer).Wait(ctx); err != nil { + return nil, false, err + } + payload, err := server.blobCluster.FetchChunkBlob(ctx, peer, ref.ContentSHA256) + if err != nil { + if status.Code(err) == codes.InvalidArgument { + server.observeS3ChunkBlobMismatch() + } + if ctx.Err() != nil { + return nil, false, errors.WithStack(ctx.Err()) + } + return nil, false, nil + } + actual := sha256.Sum256(payload) + if actual != ref.ContentSHA256 || uint64(len(payload)) != ref.Size { //nolint:gosec // Chunk size is bounded by the S3 data path. + server.observeS3ChunkBlobMismatch() + return nil, false, nil + } + return payload, true, nil +} + +func (b *S3BlobBackfiller) limiterFor(peer S3BlobReplica) *s3BlobTokenBucket { + identity := peer.NodeID + "\x00" + peer.Address + b.limitersMu.Lock() + defer b.limitersMu.Unlock() + limiter := b.limiters[identity] + if limiter == nil { + limiter = newS3BlobTokenBucket(b.config.RatePerPeer, b.config.BurstPerPeer) + b.limiters[identity] = limiter + } + return limiter +} + +func (b *S3BlobBackfiller) runScanner(ctx context.Context) { + defer b.wg.Done() + b.scanLocalChunkRefs(ctx) + ticker := time.NewTicker(b.config.ScanInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + b.scanLocalChunkRefs(ctx) + } + } +} + +func (b *S3BlobBackfiller) scanLocalChunkRefs(ctx context.Context) { + server := b.server.Load() + if server == nil { + return + } + scanner, ok := server.blobLocalStores.(S3BlobLocalStoreScanner) + if !ok { + return + } + for _, localStore := range scanner.LocalStores() { + if localStore == nil || !b.scanChunkRefsInStore(ctx, localStore) { + return + } + } +} + +func (b *S3BlobBackfiller) scanChunkRefsInStore(ctx context.Context, localStore store.MVCCStore) bool { + start := []byte(s3keys.ChunkRefPrefix) + end := prefixScanEnd(start) + for { + rows, err := localStore.ScanAt(ctx, start, end, b.config.ScanPageSize, math.MaxUint64) + if err != nil { + if ctx.Err() == nil { + slog.Warn("scan local s3 chunkrefs for backfill", "err", err) + } + return ctx.Err() == nil + } + for _, row := range rows { + if row == nil || !b.enqueue(ctx, row.Key, true) { + return false + } + } + if len(rows) < b.config.ScanPageSize { + return true + } + start = append(bytes.Clone(rows[len(rows)-1].Key), 0) + } +} + +func (b *S3BlobBackfiller) observer() S3BlobBackfillObserver { + if b == nil { + return nil + } + server := b.server.Load() + if server == nil || server.blobOffloadObserver == nil { + return nil + } + observer, _ := server.blobOffloadObserver.(S3BlobBackfillObserver) + return observer +} + +func (b *S3BlobBackfiller) observeQueueDepth() { + if observer := b.observer(); observer != nil { + observer.ObserveS3ChunkBlobBackfillQueueDepth(len(b.queue)) + } +} + +func (b *S3BlobBackfiller) observeResult(result string) { + if observer := b.observer(); observer != nil { + observer.ObserveS3ChunkBlobBackfillResult(result) + } +} + +type s3BlobTokenBucket struct { + mu sync.Mutex + rate float64 + burst float64 + tokens float64 + last time.Time +} + +func newS3BlobTokenBucket(ratePerSecond, burst int) *s3BlobTokenBucket { + return &s3BlobTokenBucket{ + rate: float64(ratePerSecond), + burst: float64(burst), + tokens: float64(burst), + last: time.Now(), + } +} + +func (b *s3BlobTokenBucket) Wait(ctx context.Context) error { + wait := b.reserve(time.Now()) + if wait <= 0 { + return nil + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-ctx.Done(): + b.cancelReservation() + return errors.WithStack(ctx.Err()) + case <-timer.C: + return nil + } +} + +func (b *s3BlobTokenBucket) cancelReservation() { + b.mu.Lock() + b.tokens = min(b.burst, b.tokens+1) + b.mu.Unlock() +} + +func (b *s3BlobTokenBucket) reserve(now time.Time) time.Duration { + b.mu.Lock() + defer b.mu.Unlock() + if now.After(b.last) { + b.tokens = min(b.burst, b.tokens+now.Sub(b.last).Seconds()*b.rate) + b.last = now + } + b.tokens-- + if b.tokens >= 0 { + return 0 + } + return time.Duration(-b.tokens / b.rate * float64(time.Second)) +} diff --git a/adapter/s3_blob_backfill_test.go b/adapter/s3_blob_backfill_test.go new file mode 100644 index 000000000..3751cd335 --- /dev/null +++ b/adapter/s3_blob_backfill_test.go @@ -0,0 +1,352 @@ +package adapter + +import ( + "bytes" + "context" + "crypto/sha256" + "math" + "slices" + "sync" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/s3keys" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestS3BlobBackfillerApplyHookDoesNotBlockOnFullQueue(t *testing.T) { + t.Parallel() + + backfiller := NewS3BlobBackfiller(S3BlobBackfillConfig{QueueSize: 1}) + first := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 0) + second := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 1) + backfiller.OnApply(pb.Op_PUT, first) + + started := time.Now() + backfiller.OnApply(pb.Op_PUT, second) + require.Less(t, time.Since(started), 100*time.Millisecond) + require.Len(t, backfiller.queue, 1) + _, firstPending := backfiller.pending.Load(string(first)) + _, secondPending := backfiller.pending.Load(string(second)) + require.True(t, firstPending) + require.False(t, secondPending) +} + +func TestS3BlobBackfillerFetchesAppliedRefFromAlternatePeerAfterMismatch(t *testing.T) { + t.Parallel() + + metadataStore := store.NewMVCCStore() + blobStore := store.NewMVCCStore() + stores := s3BlobBackfillTestStores{metadata: metadataStore, blobs: blobStore} + cluster := newFakeS3BlobCluster() + payload := []byte("backfilled payload") + digest := sha256.Sum256(payload) + cluster.blobs["n2"] = map[[sha256.Size]byte][]byte{digest: []byte("bad payload")} + cluster.blobs["n3"] = map[[sha256.Size]byte][]byte{digest: payload} + observer := &recordingS3BlobBackfillObserver{} + refKey := putS3BlobBackfillRef(t, metadataStore, digest, uint64(len(payload)), "n2") + backfiller := NewS3BlobBackfiller(s3BlobBackfillTestConfig()) + server := NewS3Server( + nil, "", metadataStore, newLocalAdapterCoordinator(metadataStore), nil, + WithS3BlobCluster(cluster), + WithS3BlobLocalStoreResolver(stores), + WithS3BlobOffloadObserver(observer), + WithS3BlobBackfiller(backfiller), + ) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { + cancel() + backfiller.Stop() + }) + require.NoError(t, server.StartBlobBackfill(ctx)) + + backfiller.OnApply(pb.Op_PUT, refKey) + require.Eventually(t, func() bool { + got, err := metadataStore.GetAt(context.Background(), s3keys.ChunkBlobKey(digest), math.MaxUint64) + return err == nil && bytes.Equal(got, payload) + }, 3*time.Second, 10*time.Millisecond) + _, err := blobStore.GetAt(context.Background(), s3keys.ChunkBlobKey(digest), math.MaxUint64) + require.ErrorIs(t, err, store.ErrKeyNotFound) + require.Equal(t, refKey, cluster.lastRouteKey()) + require.GreaterOrEqual(t, observer.shaMismatches(), 1) + require.Contains(t, observer.resultsSnapshot(), s3BlobBackfillResultFetched) +} + +func TestS3BlobBackfillerStartupScanRecoversUnobservedRef(t *testing.T) { + t.Parallel() + + metadataStore := store.NewMVCCStore() + blobStore := store.NewMVCCStore() + stores := s3BlobBackfillTestStores{metadata: metadataStore, blobs: blobStore} + cluster := newFakeS3BlobCluster() + payload := []byte("snapshot backfill") + digest := sha256.Sum256(payload) + cluster.blobs["n2"] = map[[sha256.Size]byte][]byte{digest: payload} + refKey := putS3BlobBackfillRef(t, metadataStore, digest, uint64(len(payload)), "n2") + backfiller := NewS3BlobBackfiller(s3BlobBackfillTestConfig()) + server := NewS3Server( + nil, "", metadataStore, newLocalAdapterCoordinator(metadataStore), nil, + WithS3BlobCluster(cluster), + WithS3BlobLocalStoreResolver(stores), + WithS3BlobBackfiller(backfiller), + ) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { + cancel() + backfiller.Stop() + }) + require.NoError(t, server.StartBlobBackfill(ctx)) + + require.Eventually(t, func() bool { + got, err := metadataStore.GetAt(context.Background(), s3keys.ChunkBlobKey(digest), math.MaxUint64) + return err == nil && bytes.Equal(got, payload) + }, 3*time.Second, 10*time.Millisecond) + _, err := blobStore.GetAt(context.Background(), s3keys.ChunkBlobKey(digest), math.MaxUint64) + require.ErrorIs(t, err, store.ErrKeyNotFound) + require.Equal(t, refKey, cluster.lastRouteKey()) +} + +func TestS3BlobBackfillConfigFromEnvRejectsUnboundedValues(t *testing.T) { + t.Setenv(s3BlobBackfillWorkersEnvVar, "0") + _, err := S3BlobBackfillConfigFromEnv() + require.Error(t, err) + + t.Setenv(s3BlobBackfillWorkersEnvVar, "2") + t.Setenv(s3BlobBackfillScanIntervalEnvVar, "not-a-duration") + _, err = S3BlobBackfillConfigFromEnv() + require.Error(t, err) +} + +func TestS3BlobTokenBucketAllowsBurstThenRefills(t *testing.T) { + t.Parallel() + + now := time.Unix(100, 0) + bucket := &s3BlobTokenBucket{rate: 2, burst: 2, tokens: 2, last: now} + require.Zero(t, bucket.reserve(now)) + require.Zero(t, bucket.reserve(now)) + require.Equal(t, 500*time.Millisecond, bucket.reserve(now)) + require.Equal(t, 750*time.Millisecond, bucket.reserve(now.Add(250*time.Millisecond))) + require.Equal(t, time.Second, bucket.reserve(now.Add(500*time.Millisecond))) +} + +func TestS3BlobTokenBucketReservesDistinctConcurrentSlots(t *testing.T) { + t.Parallel() + + now := time.Unix(100, 0) + bucket := &s3BlobTokenBucket{rate: 1, burst: 1, tokens: 1, last: now} + waits := make(chan time.Duration, 4) + var wg sync.WaitGroup + for range 4 { + wg.Add(1) + go func() { + defer wg.Done() + waits <- bucket.reserve(now) + }() + } + wg.Wait() + close(waits) + got := make([]time.Duration, 0, 4) + for wait := range waits { + got = append(got, wait) + } + slices.Sort(got) + require.Equal(t, []time.Duration{0, time.Second, 2 * time.Second, 3 * time.Second}, got) +} + +func TestS3BlobBackfillerStopClearsQueuedPendingKeys(t *testing.T) { + t.Parallel() + + backfiller := NewS3BlobBackfiller(S3BlobBackfillConfig{QueueSize: 2}) + key := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 0) + backfiller.OnApply(pb.Op_PUT, key) + backfiller.Stop() + + require.Empty(t, backfiller.queue) + _, pending := backfiller.pending.Load(string(key)) + require.False(t, pending) + backfiller.OnApply(pb.Op_PUT, key) + require.Len(t, backfiller.queue, 1) +} + +func TestS3BlobBackfillerApplyWhileStartingPublishesServerSafely(t *testing.T) { + t.Parallel() + + metadataStore := store.NewMVCCStore() + backfiller := NewS3BlobBackfiller(s3BlobBackfillTestConfig()) + server := NewS3Server( + nil, "", metadataStore, newLocalAdapterCoordinator(metadataStore), nil, + WithS3BlobCluster(newFakeS3BlobCluster()), + WithS3BlobLocalStoreResolver(s3BlobBackfillTestStores{metadata: metadataStore}), + WithS3BlobBackfiller(backfiller), + ) + key := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 0) + done := make(chan struct{}) + go func() { + defer close(done) + for range 100 { + backfiller.OnApply(pb.Op_PUT, key) + } + }() + require.NoError(t, server.StartBlobBackfill(context.Background())) + <-done + backfiller.Stop() +} + +func TestStoreFetchedS3ChunkBlobUsesReplicatedRefTimestamp(t *testing.T) { + t.Parallel() + + local := &recordingS3BlobFetchStore{MVCCStore: store.NewMVCCStore()} + server := &S3Server{blobLocalStores: &mutableS3BlobLocalStore{store: local}} + refKey := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 0) + payload := []byte("follower-safe repair") + digest := sha256.Sum256(payload) + + require.NoError(t, server.storeFetchedS3ChunkBlob(context.Background(), refKey, digest, payload, 42)) + require.Equal(t, []uint64{42}, local.applyCommitTS) +} + +func TestStoreFetchedS3ChunkBlobRepairsCorruptNewerVersion(t *testing.T) { + t.Parallel() + + base := store.NewMVCCStore() + payload := []byte("verified repair payload") + digest := sha256.Sum256(payload) + key := s3keys.ChunkBlobKey(digest) + require.NoError(t, base.PutAt(context.Background(), key, []byte("corrupt"), 100, 0)) + local := &recordingS3BlobFetchStore{MVCCStore: base} + server := &S3Server{blobLocalStores: &mutableS3BlobLocalStore{store: local}} + refKey := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 0) + + require.NoError(t, server.storeFetchedS3ChunkBlob(context.Background(), refKey, digest, payload, 42)) + require.Equal(t, []uint64{100}, local.applyStartTS) + require.Equal(t, []uint64{101}, local.applyCommitTS) + got, err := base.GetAt(context.Background(), key, math.MaxUint64) + require.NoError(t, err) + require.Equal(t, payload, got) +} + +func TestS3BlobBackfillerRepairsCorruptLocalBlob(t *testing.T) { + t.Parallel() + + metadataStore := store.NewMVCCStore() + payload := []byte("background repair payload") + digest := sha256.Sum256(payload) + key := s3keys.ChunkBlobKey(digest) + require.NoError(t, metadataStore.PutAt(context.Background(), key, []byte("corrupt"), 20, 0)) + cluster := newFakeS3BlobCluster() + cluster.blobs["n2"] = map[[sha256.Size]byte][]byte{digest: payload} + putS3BlobBackfillRef(t, metadataStore, digest, uint64(len(payload)), "n2") + backfiller := NewS3BlobBackfiller(s3BlobBackfillTestConfig()) + server := NewS3Server( + nil, "", metadataStore, newLocalAdapterCoordinator(metadataStore), nil, + WithS3BlobCluster(cluster), + WithS3BlobLocalStoreResolver(s3BlobBackfillTestStores{metadata: metadataStore}), + WithS3BlobBackfiller(backfiller), + ) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { + cancel() + backfiller.Stop() + }) + require.NoError(t, server.StartBlobBackfill(ctx)) + + require.Eventually(t, func() bool { + got, err := metadataStore.GetAt(context.Background(), key, math.MaxUint64) + return err == nil && bytes.Equal(got, payload) + }, 3*time.Second, 10*time.Millisecond) + latestTS, exists, err := metadataStore.LatestCommitTS(context.Background(), key) + require.NoError(t, err) + require.True(t, exists) + require.Equal(t, uint64(21), latestTS) +} + +func s3BlobBackfillTestConfig() S3BlobBackfillConfig { + return S3BlobBackfillConfig{ + Workers: 1, + QueueSize: 8, + RatePerPeer: 1000, + BurstPerPeer: 8, + ScanInterval: time.Hour, + ScanPageSize: 2, + MaxAttempts: 1, + RetryInitial: time.Millisecond, + RetryMax: time.Millisecond, + } +} + +func putS3BlobBackfillRef( + t *testing.T, + st store.MVCCStore, + digest [sha256.Size]byte, + size uint64, + source string, +) []byte { + t.Helper() + key := s3keys.ChunkRefKey("bucket", 1, "object", "upload", 1, 0) + value, err := s3keys.EncodeChunkRefValue(s3keys.ChunkRefValue{ + ContentSHA256: digest, + Size: size, + SourcePeer: source, + ReplicaPeers: []s3keys.ChunkRefPeer{ + {NodeID: "n2", Address: "n2:50051"}, + {NodeID: "n3", Address: "n3:50051"}, + }, + }) + require.NoError(t, err) + require.NoError(t, st.PutAt(context.Background(), key, value, 10, 0)) + return key +} + +type s3BlobBackfillTestStores struct { + metadata store.MVCCStore + blobs store.MVCCStore +} + +func (s s3BlobBackfillTestStores) LocalStoreForKey(key []byte) (store.MVCCStore, bool) { + if bytes.HasPrefix(key, []byte(s3keys.ChunkBlobPrefix)) { + return s.blobs, s.blobs != nil + } + return s.metadata, s.metadata != nil +} + +func (s s3BlobBackfillTestStores) LocalStores() []store.MVCCStore { + return []store.MVCCStore{s.metadata, s.blobs} +} + +type recordingS3BlobBackfillObserver struct { + mu sync.Mutex + mismatch int + results []string +} + +func (o *recordingS3BlobBackfillObserver) ObserveS3BlobOffloadDecision(string, string) {} +func (o *recordingS3BlobBackfillObserver) ObserveS3ChunkBlobReplicationDegraded() {} +func (o *recordingS3BlobBackfillObserver) ObserveS3ChunkBlobUnrecoverable() {} + +func (o *recordingS3BlobBackfillObserver) ObserveS3ChunkBlobSHAMismatch() { + o.mu.Lock() + o.mismatch++ + o.mu.Unlock() +} + +func (o *recordingS3BlobBackfillObserver) ObserveS3ChunkBlobBackfillQueueDepth(int) {} + +func (o *recordingS3BlobBackfillObserver) ObserveS3ChunkBlobBackfillResult(result string) { + o.mu.Lock() + o.results = append(o.results, result) + o.mu.Unlock() +} + +func (o *recordingS3BlobBackfillObserver) shaMismatches() int { + o.mu.Lock() + defer o.mu.Unlock() + return o.mismatch +} + +func (o *recordingS3BlobBackfillObserver) resultsSnapshot() []string { + o.mu.Lock() + defer o.mu.Unlock() + return append([]string(nil), o.results...) +} diff --git a/adapter/s3_blob_fetch.go b/adapter/s3_blob_fetch.go index 452f18b73..b499cd8b6 100644 --- a/adapter/s3_blob_fetch.go +++ b/adapter/s3_blob_fetch.go @@ -81,7 +81,7 @@ func (s *S3BlobFetchServer) FetchChunkBlob(req *pb.FetchChunkBlobRequest, stream if err != nil { return err } - if err := s.verifyChunkBlobDigest(digest, payload, codes.InvalidArgument); err != nil { + if err := s.verifyChunkBlobDigest(digest, payload); err != nil { return err } return sendChunkBlobPayload(stream, payload) @@ -130,7 +130,7 @@ func (s *S3BlobFetchServer) PushChunkBlob(stream pb.S3BlobFetch_PushChunkBlobSer if err != nil { return err } - if err := s.verifyChunkBlobDigest(digest, payload, codes.InvalidArgument); err != nil { + if err := s.verifyChunkBlobDigest(digest, payload); err != nil { return err } if err := s.ensurePushAllowed(); err != nil { @@ -183,6 +183,80 @@ func (s *S3BlobFetchServer) storeChunkBlob( } } +// repairChunkBlob stores a peer-verified payload without requiring a follower +// to allocate a cluster timestamp. A corrupt local version may already be at +// or beyond the replicated chunkref timestamp, so repairs advance only this +// peer-local key while preserving the store-wide MVCC watermark. +func (s *S3BlobFetchServer) repairChunkBlob( + ctx context.Context, + digest [s3ChunkBlobSHA256Bytes]byte, + payload []byte, + minCommitTS uint64, +) error { + if minCommitTS == 0 { + return s3BlobFetchStatus(codes.InvalidArgument, "missing s3 chunkblob repair timestamp") + } + if err := s.verifyChunkBlobDigest(digest, payload); err != nil { + return err + } + key := s3keys.ChunkBlobKey(digest) + for { + if err := s.ensurePushAllowed(); err != nil { + return err + } + startTS, repairTS, done, err := s.chunkBlobRepairVersion(ctx, key, payload, minCommitTS) + if err != nil { + return err + } + if done { + return nil + } + if err := s.applyChunkBlobUntilRegistered(ctx, key, payload, startTS, repairTS); err != nil { + retry, repairErr := normalizeS3ChunkBlobRepairError(err) + if retry { + continue + } + return repairErr + } + return nil + } +} + +func (s *S3BlobFetchServer) chunkBlobRepairVersion( + ctx context.Context, + key, payload []byte, + minCommitTS uint64, +) (uint64, uint64, bool, error) { + latestTS, exists, err := s.latestChunkBlobTS(ctx, key) + if err != nil { + return 0, 0, false, err + } + existing, currentExists, err := s.currentChunkBlobPayload(ctx, key) + if err != nil { + return 0, 0, false, err + } + if currentExists && bytes.Equal(existing, payload) { + return 0, 0, true, nil + } + if !exists || minCommitTS > latestTS { + return latestTS, minCommitTS, false, nil + } + if latestTS == math.MaxUint64 { + return 0, 0, false, s3BlobFetchStatus(codes.FailedPrecondition, "s3 chunkblob repair timestamp is exhausted") + } + return latestTS, latestTS + 1, false, nil +} + +func normalizeS3ChunkBlobRepairError(err error) (bool, error) { + if errors.Is(err, store.ErrWriteConflict) { + return true, nil + } + if code := status.Code(err); code != codes.Unknown { + return false, err + } + return false, s3BlobFetchStatusf(codes.Internal, "repair s3 chunkblob: %v", err) +} + func (s *S3BlobFetchServer) retryAfterChunkBlobWriteConflict( ctx context.Context, err error, @@ -244,7 +318,7 @@ func (s *S3BlobFetchServer) chunkBlobAlreadyStored( if bytes.Equal(existing, payload) { return true, nil } - if err := s.verifyChunkBlobDigest(digest, existing, codes.InvalidArgument); err != nil { + if err := s.verifyChunkBlobDigest(digest, existing); err != nil { return false, err } return false, s3BlobFetchStatus(codes.InvalidArgument, "s3 chunkblob already exists with different payload") @@ -440,13 +514,13 @@ func (s *s3ChunkBlobReceiveState) applyDigest(raw []byte) error { return nil } -func (s *S3BlobFetchServer) verifyChunkBlobDigest(expected [s3ChunkBlobSHA256Bytes]byte, payload []byte, code codes.Code) error { +func (s *S3BlobFetchServer) verifyChunkBlobDigest(expected [s3ChunkBlobSHA256Bytes]byte, payload []byte) error { actual := sha256.Sum256(payload) if actual == expected { return nil } s.observeSHAMismatch() - return s3BlobFetchStatus(code, "s3 chunkblob sha256 mismatch") + return s3BlobFetchStatus(codes.InvalidArgument, "s3 chunkblob sha256 mismatch") } func s3ChunkBlobDigest(raw []byte) ([s3ChunkBlobSHA256Bytes]byte, error) { diff --git a/adapter/s3_blob_offload.go b/adapter/s3_blob_offload.go index 88ccfd1b9..707d84291 100644 --- a/adapter/s3_blob_offload.go +++ b/adapter/s3_blob_offload.go @@ -122,6 +122,15 @@ func WithS3BlobPushBlocked(blocked func() bool) S3ServerOption { } } +func WithS3BlobBackfiller(backfiller *S3BlobBackfiller) S3ServerOption { + return func(server *S3Server) { + if server == nil { + return + } + server.blobBackfiller = backfiller + } +} + // S3BlobMinReplicasFromEnv returns zero when the dynamic Raft-quorum default // should be used. Explicit values below two are rejected because a leader-only // chunkblob is weaker than the legacy Raft data path. diff --git a/adapter/s3_blob_read.go b/adapter/s3_blob_read.go index 0bd21aeaf..3067ff433 100644 --- a/adapter/s3_blob_read.go +++ b/adapter/s3_blob_read.go @@ -185,6 +185,11 @@ func (s *S3Server) localS3ChunkBlob(ctx context.Context, routeKey []byte, digest return payload, true, nil } +func (s *S3Server) localS3ChunkBlobExists(ctx context.Context, routeKey []byte, digest [s3ChunkBlobSHA256Bytes]byte) (bool, error) { + _, exists, err := s.localS3ChunkBlob(ctx, routeKey, digest) + return exists, err +} + func (s *S3Server) fetchS3ChunkBlob(ctx context.Context, routeKey []byte, ref s3keys.ChunkRefValue) ([]byte, error) { if s == nil || s.blobCluster == nil { return nil, s3BlobUnavailable("s3 chunkblob peer client is not configured") @@ -281,9 +286,8 @@ func (s *S3Server) storeFetchedS3ChunkBlob(ctx context.Context, routeKey []byte, if !ok || localStore == nil { return s3BlobUnavailable("s3 chunkblob local store is unavailable") } - repairTS, err := s.nextTxnCommitTS(ctx, commitTS) - if err != nil { - return errors.Wrap(err, "allocate s3 chunkblob repair timestamp") + if commitTS == 0 { + return s3BlobUnavailable("s3 chunkblob repair commit timestamp is unavailable") } server := NewS3BlobFetchServer( localStore, @@ -291,7 +295,7 @@ func (s *S3Server) storeFetchedS3ChunkBlob(ctx context.Context, routeKey []byte, WithS3BlobFetchClock(s.clock()), WithS3BlobFetchPushBlocked(s.blobPushBlocked), ) - return server.storeChunkBlob(ctx, digest, payload, repairTS) + return server.repairChunkBlob(ctx, digest, payload, commitTS) } func (s *S3Server) observeS3ChunkBlobMismatch() { diff --git a/kv/shard_store.go b/kv/shard_store.go index 4d071dd35..fc3276de9 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -2427,6 +2427,25 @@ func (s *ShardStore) LocalStoreForKey(key []byte) (store.MVCCStore, bool) { return g.Store, true } +// LocalStores returns every process-local shard store in stable group order. +// It is used by node-local auxiliary maintenance that must recover state after +// snapshot restore without leader routing. +func (s *ShardStore) LocalStores() []store.MVCCStore { + groupIDs := make([]uint64, 0, len(s.groups)) + for groupID := range s.groups { + groupIDs = append(groupIDs, groupID) + } + sort.Slice(groupIDs, func(i, j int) bool { return groupIDs[i] < groupIDs[j] }) + stores := make([]store.MVCCStore, 0, len(groupIDs)) + for _, groupID := range groupIDs { + group := s.groups[groupID] + if group != nil && group.Store != nil { + stores = append(stores, group.Store) + } + } + return stores +} + func (s *ShardStore) proxyRawGet(ctx context.Context, g *ShardGroup, key []byte, ts uint64, groupID uint64) ([]byte, error) { engine := engineForGroup(g) if engine == nil { diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index 5587661a6..83f6ceb63 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -118,6 +118,20 @@ func TestShardStoreScanAt_RoutesListItemScansByUserKey(t *testing.T) { require.Equal(t, k2, kvs[2].Key) } +func TestShardStoreLocalStoresUsesStableGroupOrder(t *testing.T) { + t.Parallel() + + first := store.NewMVCCStore() + second := store.NewMVCCStore() + shards := NewShardStore(distribution.NewEngine(), map[uint64]*ShardGroup{ + 20: {Store: second}, + 10: {Store: first}, + 30: nil, + }) + + require.Equal(t, []store.MVCCStore{first, second}, shards.LocalStores()) +} + func TestShardStoreScanGroupAt_UsesExplicitGroup(t *testing.T) { t.Parallel() diff --git a/main.go b/main.go index 5bc27a830..15f6fe168 100644 --- a/main.go +++ b/main.go @@ -447,6 +447,7 @@ func run() error { } keystore := encryption.NewKeystore() redisApplyObserver := adapter.NewRedisApplyObserver() + s3BlobBackfiller := adapter.NewS3BlobBackfiller(cfg.s3BlobBackfillConfig) // Stage 6D-6c: buildShardGroupsWithEncryptionWiring assembles the // storage-envelope write-path wiring (cipher + deterministic nonce @@ -475,6 +476,7 @@ func run() error { *encryptionEnabled, cfg.engine, redisApplyObserver, + s3BlobBackfiller, ) if err = chainEncryptionStartupGuard( err, @@ -599,6 +601,7 @@ func run() error { metricsRegistry: metricsRegistry, cfg: cfg, redisApplyObserver: redisApplyObserver, cleanup: &cleanup, + s3BlobBackfiller: s3BlobBackfiller, encWiring: encWiring, keyvizSampler: sampler, encryptionConfChangeInterceptor: encryptionConfChangeInterceptor, @@ -862,6 +865,10 @@ func resolveRuntimeInputs() (runtimeConfig, raftEngineType, raftBootstrapConfig, if err != nil { return runtimeConfig{}, "", raftBootstrapConfig{}, false, err } + cfg.s3BlobBackfillConfig, err = adapter.S3BlobBackfillConfigFromEnv() + if err != nil { + return runtimeConfig{}, "", raftBootstrapConfig{}, false, errors.WithStack(err) + } bootstrapCfg, err := resolveRaftPeerConfig( *raftId, @@ -931,15 +938,16 @@ func cloneRaftServers(in []raftengine.Server) []raftengine.Server { } type runtimeConfig struct { - groups []groupSpec - defaultGroup uint64 - engine *distribution.Engine - leaderRedis map[string]string - leaderS3 map[string]string - leaderDynamo map[string]string - leaderSQS map[string]string - sqsFifoPartitionMap map[string]sqsFifoQueueRouting - multi bool + groups []groupSpec + defaultGroup uint64 + engine *distribution.Engine + leaderRedis map[string]string + leaderS3 map[string]string + leaderDynamo map[string]string + leaderSQS map[string]string + sqsFifoPartitionMap map[string]sqsFifoQueueRouting + s3BlobBackfillConfig adapter.S3BlobBackfillConfig + multi bool } func parseRuntimeConfig(myAddr, redisAddr, s3Addr, dynamoAddr, sqsAddr, raftGroups, shardRanges, raftRedisMap, raftS3Map, raftDynamoMap, raftSqsMap, sqsFifoPartitionMapRaw string) (runtimeConfig, error) { @@ -1775,6 +1783,7 @@ type serversInput struct { encWiring encryptionWriteWiring redisApplyObserver *adapter.RedisApplyObserver cleanup *internalutil.CleanupStack + s3BlobBackfiller *adapter.S3BlobBackfiller // keyvizSampler is the in-memory key visualizer sampler, or nil // when --keyvizEnabled is false. Threaded into setupAdminService // so AdminServer.GetKeyVizMatrix can serve snapshots; the @@ -1857,6 +1866,7 @@ func startServersAfterStartupRotation(waitRotateOnStartup startupRotationWaiter, readTracker: in.readTracker, encWiring: in.encWiring, redisApplyObserver: in.redisApplyObserver, + s3BlobBackfiller: in.s3BlobBackfiller, dynamoAddress: *dynamoAddr, leaderDynamo: in.cfg.leaderDynamo, s3Address: *s3Addr, @@ -3010,6 +3020,7 @@ type runtimeServerRunner struct { pubsubRelay *adapter.RedisPubSubRelay readTracker *kv.ActiveTimestampTracker redisApplyObserver *adapter.RedisApplyObserver + s3BlobBackfiller *adapter.S3BlobBackfiller encWiring encryptionWriteWiring dynamoAddress string leaderDynamo map[string]string @@ -3151,6 +3162,7 @@ func (r *runtimeServerRunner) prepareAdminForwardServers() error { r.metricsRegistry.S3BlobOffloadObserver(), blobCluster, r.publicKVGate.blocked, + r.s3BlobBackfiller, ) if err != nil { if blobCluster != nil { @@ -3163,7 +3175,7 @@ func (r *runtimeServerRunner) prepareAdminForwardServers() error { } func (r *runtimeServerRunner) newS3BlobCluster() (adapter.S3BlobCluster, error) { - if r == nil || strings.TrimSpace(r.s3Address) == "" { + if r == nil || !s3BlobNodeEnabled(r.s3Address, *s3BlobPeerTokenFile) { return nil, nil } members, ok := r.coordinate.(kv.RaftMembershipCoordinator) @@ -3294,8 +3306,12 @@ func (r *runtimeServerRunner) startPublicServices() error { r.redisListener = nil runDynamoDBServer(r.ctx, r.eg, r.dynamoServer) r.dynamoListener = nil - runS3Server(r.ctx, r.eg, r.s3Server) - r.s3Listener = nil + if r.s3Listener != nil { + runS3Server(r.ctx, r.eg, r.s3Server) + r.s3Listener = nil + } else { + runS3BlobBackfillOnly(r.ctx, r.eg, r.s3Server) + } runSQSServer(r.ctx, r.eg, r.sqsServer) r.sqsListener = nil // Plug the SQS adapter into the monitoring registry's depth diff --git a/main_encryption_write_wiring.go b/main_encryption_write_wiring.go index b3f124e71..4068433e2 100644 --- a/main_encryption_write_wiring.go +++ b/main_encryption_write_wiring.go @@ -37,7 +37,7 @@ func buildShardGroupsWithEncryptionWiring( sidecarPath string, encryptionEnabled bool, routeEngine *distribution.Engine, - applyObserver kv.ApplyObserver, + applyObservers ...kv.ApplyObserver, ) ([]*raftGroupRuntime, map[uint64]*kv.ShardGroup, encryptionWriteWiring, error) { if guardErr := checkEncryptionMembershipStartupGuardsBeforeEngine(encryptionMembershipStartupGuardInput{ raftID: raftID, @@ -57,7 +57,7 @@ func buildShardGroupsWithEncryptionWiring( } configureRaftEnvelopeFactory(factory, encWiring) runtimes, shardGroups, err := buildShardGroups(raftID, raftDir, groups, multi, bootstrap, bootstrapCfg, - factory, proposalObserverForGroup, clock, kekWrapper, keystore, sidecarPath, encWiring, routeEngine, applyObserver) + factory, proposalObserverForGroup, clock, kekWrapper, keystore, sidecarPath, encWiring, routeEngine, applyObservers...) if err != nil { return runtimes, shardGroups, encWiring, err } diff --git a/main_s3.go b/main_s3.go index e80aedfba..0b916688d 100644 --- a/main_s3.go +++ b/main_s3.go @@ -38,7 +38,7 @@ func startS3Server( s3Server, _, err := prepareS3Server( ctx, lc, s3Addr, shardStore, coordinate, leaderS3, region, credentialsFile, pathStyleOnly, readTracker, putAdmissionObserver, - blobOffloadObserver, + blobOffloadObserver, nil, ) if err != nil { return nil, err @@ -60,10 +60,11 @@ func prepareS3Server( readTracker *kv.ActiveTimestampTracker, putAdmissionObserver adapter.S3PutAdmissionObserver, blobOffloadObserver adapter.S3BlobOffloadObserver, + blobBackfiller *adapter.S3BlobBackfiller, ) (*adapter.S3Server, net.Listener, error) { s3Server, err := newS3Server( s3Addr, shardStore, coordinate, leaderS3, region, credentialsFile, - pathStyleOnly, readTracker, putAdmissionObserver, blobOffloadObserver, nil, nil, + pathStyleOnly, readTracker, putAdmissionObserver, blobOffloadObserver, nil, nil, blobBackfiller, ) if err != nil { return nil, nil, err @@ -88,19 +89,21 @@ func newS3Server( blobOffloadObserver adapter.S3BlobOffloadObserver, blobCluster adapter.S3BlobCluster, blobPushBlocked func() bool, + blobBackfiller *adapter.S3BlobBackfiller, ) (*adapter.S3Server, error) { s3Addr = strings.TrimSpace(s3Addr) - if s3Addr == "" { + peerOnly := s3Addr == "" && blobCluster != nil && blobBackfiller != nil + if s3Addr == "" && !peerOnly { // (nil, nil) is the explicit "S3 disabled" signal — the empty // flag value is a valid configuration, not an error. The // nilnil linter is not enabled in .golangci.yaml so no // suppression directive is needed. return nil, nil } - if !pathStyleOnly { + if s3Addr != "" && !pathStyleOnly { return nil, errors.New("virtual-hosted style S3 requests are not implemented") } - staticCreds, err := loadS3StaticCredentials(credentialsFile) + staticCreds, err := loadS3StaticCredentialsForAddress(s3Addr, credentialsFile) if err != nil { return nil, err } @@ -116,6 +119,7 @@ func newS3Server( adapter.WithS3BlobOffloadObserver(blobOffloadObserver), adapter.WithS3BlobMinReplicas(minReplicas), adapter.WithS3BlobPushBlocked(blobPushBlocked), + adapter.WithS3BlobBackfiller(blobBackfiller), } if blobCluster != nil { options = append(options, adapter.WithS3BlobCluster(blobCluster)) @@ -158,6 +162,11 @@ func runS3Server(ctx context.Context, eg *errgroup.Group, s3Server *adapter.S3Se return nil }) eg.Go(func() error { + if err := s3Server.StartBlobBackfill(ctx); err != nil { + s3Server.Stop() + runDoneCancel() + return errors.WithStack(err) + } err := s3Server.Run() runDoneCancel() if err == nil || errors.Is(err, net.ErrClosed) { @@ -167,6 +176,32 @@ func runS3Server(ctx context.Context, eg *errgroup.Group, s3Server *adapter.S3Se }) } +func runS3BlobBackfillOnly(ctx context.Context, eg *errgroup.Group, s3Server *adapter.S3Server) { + if s3Server == nil { + return + } + eg.Go(func() error { + if err := s3Server.StartBlobBackfill(ctx); err != nil { + s3Server.Stop() + return errors.WithStack(err) + } + <-ctx.Done() + s3Server.Stop() + return nil + }) +} + +func s3BlobNodeEnabled(s3Address, peerTokenFile string) bool { + return strings.TrimSpace(s3Address) != "" || strings.TrimSpace(peerTokenFile) != "" +} + +func loadS3StaticCredentialsForAddress(s3Address, credentialsFile string) (map[string]string, error) { + if s3Address == "" { + return nil, nil + } + return loadS3StaticCredentials(credentialsFile) +} + func loadS3StaticCredentials(path string) (map[string]string, error) { return loadSigV4StaticCredentialsFile(path, "s3") } diff --git a/main_s3_test.go b/main_s3_test.go index 162091112..456268920 100644 --- a/main_s3_test.go +++ b/main_s3_test.go @@ -2,11 +2,17 @@ package main import ( "context" + "crypto/sha256" "net" "os" "path/filepath" "testing" + "time" + "github.com/bootjp/elastickv/adapter" + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" ) @@ -25,6 +31,56 @@ func TestStartS3ServerAllowsEmptyAddress(t *testing.T) { require.Nil(t, srv) } +func TestS3BlobNodeEnabledForPeerOnlyNode(t *testing.T) { + t.Parallel() + + require.True(t, s3BlobNodeEnabled("", "/run/secrets/s3-peer-token")) + require.True(t, s3BlobNodeEnabled("127.0.0.1:9000", "")) + require.False(t, s3BlobNodeEnabled("", "")) +} + +func TestRunS3BlobBackfillOnlyStartsListenlessServer(t *testing.T) { + t.Parallel() + + engine := distribution.NewEngine() + engine.UpdateRoute(nil, nil, 1) + shardStore := kv.NewShardStore(engine, map[uint64]*kv.ShardGroup{ + 1: {Store: store.NewMVCCStore()}, + }) + backfiller := adapter.NewS3BlobBackfiller(adapter.S3BlobBackfillConfig{ + Workers: 1, QueueSize: 1, RatePerPeer: 1, BurstPerPeer: 1, + ScanInterval: time.Hour, ScanPageSize: 1, MaxAttempts: 1, + RetryInitial: time.Millisecond, RetryMax: time.Millisecond, + }) + server, err := newS3Server( + "", shardStore, &stubStartupCoordinator{clock: kv.NewHLC()}, nil, "", "", false, nil, + nil, nil, peerOnlyS3BlobCluster{}, nil, backfiller, + ) + require.NoError(t, err) + require.NotNil(t, server) + + ctx, cancel := context.WithCancel(context.Background()) + eg, runCtx := errgroup.WithContext(ctx) + runS3BlobBackfillOnly(runCtx, eg, server) + cancel() + require.NoError(t, eg.Wait()) +} + +type peerOnlyS3BlobCluster struct{} + +func (peerOnlyS3BlobCluster) AllPeersSupportS3BlobOffload(context.Context) bool { return true } +func (peerOnlyS3BlobCluster) SelfNodeID() string { return "n1" } +func (peerOnlyS3BlobCluster) ReplicasForChunk(context.Context, []byte) ([]adapter.S3BlobReplica, error) { + return nil, nil +} +func (peerOnlyS3BlobCluster) PushChunkBlob(context.Context, adapter.S3BlobReplica, [sha256.Size]byte, []byte, uint64) error { + return nil +} +func (peerOnlyS3BlobCluster) FetchChunkBlob(context.Context, adapter.S3BlobReplica, [sha256.Size]byte) ([]byte, error) { + return nil, nil +} +func (peerOnlyS3BlobCluster) Close() error { return nil } + func TestLoadS3StaticCredentials(t *testing.T) { path := filepath.Join(t.TempDir(), "s3creds.json") err := os.WriteFile(path, []byte(`{"credentials":[{"access_key_id":"akid","secret_access_key":"secret"}]}`), 0o600) diff --git a/monitoring/s3.go b/monitoring/s3.go index cdc74def7..8757f27da 100644 --- a/monitoring/s3.go +++ b/monitoring/s3.go @@ -47,6 +47,8 @@ type S3Metrics struct { chunkBlobReplicationDegraded prometheus.Counter chunkBlobSHAMismatches prometheus.Counter chunkBlobUnrecoverableReads prometheus.Counter + chunkBlobBackfillQueueDepth prometheus.Gauge + chunkBlobBackfillResults *prometheus.CounterVec } func newS3Metrics(registerer prometheus.Registerer) *S3Metrics { @@ -97,6 +99,19 @@ func newS3Metrics(registerer prometheus.Registerer) *S3Metrics { Help: "Total S3 chunkblob reads that could not recover the referenced content from any available peer.", }, ), + chunkBlobBackfillQueueDepth: prometheus.NewGauge( + prometheus.GaugeOpts{ + Name: "elastickv_s3_chunkblob_backfill_queue_depth", + Help: "Current number of S3 chunkrefs queued for asynchronous local blob backfill.", + }, + ), + chunkBlobBackfillResults: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_s3_chunkblob_backfill_results_total", + Help: "Total S3 chunkblob backfill outcomes.", + }, + []string{"result"}, + ), } registerer.MustRegister( m.putAdmissionInflightBytes, @@ -106,6 +121,8 @@ func newS3Metrics(registerer prometheus.Registerer) *S3Metrics { m.chunkBlobReplicationDegraded, m.chunkBlobSHAMismatches, m.chunkBlobUnrecoverableReads, + m.chunkBlobBackfillQueueDepth, + m.chunkBlobBackfillResults, ) return m } @@ -171,6 +188,29 @@ func (m *S3Metrics) ObserveS3ChunkBlobUnrecoverable() { m.chunkBlobUnrecoverableReads.Inc() } +func (m *S3Metrics) ObserveS3ChunkBlobBackfillQueueDepth(depth int) { + if m == nil { + return + } + m.chunkBlobBackfillQueueDepth.Set(float64(max(0, depth))) +} + +func (m *S3Metrics) ObserveS3ChunkBlobBackfillResult(result string) { + if m == nil { + return + } + m.chunkBlobBackfillResults.WithLabelValues(normalizeS3BlobBackfillResult(result)).Inc() +} + +func normalizeS3BlobBackfillResult(result string) string { + switch result { + case "fetched", "local_hit", "gone", "failed", "queue_drop": + return result + default: + return s3BlobOffloadModeUnknown + } +} + func normalizeS3PutAdmissionStage(stage string) string { switch stage { case s3PutAdmissionStagePrereserve, s3PutAdmissionStagePerBatch: diff --git a/monitoring/s3_test.go b/monitoring/s3_test.go index 010ca560c..035c0df7f 100644 --- a/monitoring/s3_test.go +++ b/monitoring/s3_test.go @@ -50,6 +50,9 @@ func TestS3BlobOffloadMetricsObserve(t *testing.T) { metrics.ObserveS3ChunkBlobReplicationDegraded() metrics.ObserveS3ChunkBlobSHAMismatch() metrics.ObserveS3ChunkBlobUnrecoverable() + metrics.ObserveS3ChunkBlobBackfillQueueDepth(7) + metrics.ObserveS3ChunkBlobBackfillResult("fetched") + metrics.ObserveS3ChunkBlobBackfillResult("bogus") err := testutil.GatherAndCompare( reg, @@ -67,11 +70,20 @@ elastickv_s3_chunkblob_sha_mismatch_total 1 # HELP elastickv_s3_chunkblob_unrecoverable_total Total S3 chunkblob reads that could not recover the referenced content from any available peer. # TYPE elastickv_s3_chunkblob_unrecoverable_total counter elastickv_s3_chunkblob_unrecoverable_total 1 +# HELP elastickv_s3_chunkblob_backfill_queue_depth Current number of S3 chunkrefs queued for asynchronous local blob backfill. +# TYPE elastickv_s3_chunkblob_backfill_queue_depth gauge +elastickv_s3_chunkblob_backfill_queue_depth 7 +# HELP elastickv_s3_chunkblob_backfill_results_total Total S3 chunkblob backfill outcomes. +# TYPE elastickv_s3_chunkblob_backfill_results_total counter +elastickv_s3_chunkblob_backfill_results_total{result="fetched"} 1 +elastickv_s3_chunkblob_backfill_results_total{result="unknown"} 1 `), "elastickv_s3_blob_offload_write_decisions_total", "elastickv_s3_chunkblob_replication_degraded_total", "elastickv_s3_chunkblob_sha_mismatch_total", "elastickv_s3_chunkblob_unrecoverable_total", + "elastickv_s3_chunkblob_backfill_queue_depth", + "elastickv_s3_chunkblob_backfill_results_total", ) require.NoError(t, err) }