diff --git a/lib/backend/backend.go b/lib/backend/backend.go index 1383ed0f3d5..ea6d40be5c9 100644 --- a/lib/backend/backend.go +++ b/lib/backend/backend.go @@ -30,6 +30,7 @@ import ( "github.com/jonboulle/clockwork" "github.com/gravitational/teleport/api/types" + "github.com/gravitational/teleport/lib/utils/slices" ) // Forever means that object TTL will not expire unless deleted @@ -385,17 +386,32 @@ type BatchPutter interface { PutBatch(context.Context, []Item) ([]string, error) } +// BatchDeleter is an optional interface that backends can implement +// to support batched DeleteBatch operations for improved performance when +// deleting multiple items at once. +type BatchDeleter interface { + // DeleteBatch deletes multiple keys from the backend in a single call, in a + // way that is equivalent to a loop around multiple invocations of + // [backend.Delete], but with the potential to be more efficient or faster, + // depending on the implementation. Keys that do not exist are silently + // ignored. If an error is returned, it's possible for some of the keys to + // have already been deleted. + DeleteBatch(context.Context, []Key) error +} + // PutBatch is an implementation of PutBatch that by default calls Put for each item. // Backends can overwrite this behavior providing optimized PutBatch implementation. // // WARNING: Make sure that items have unique keys when calling PutBatch. +// +// TODO(smallinsky): Move to backend interfaces and make required for backends to +// implement batch deletes interfaces. func PutBatch(ctx context.Context, bk Backend, items []Item) ([]string, error) { if v, hasDuplicate := hasDuplicateKeys(items); hasDuplicate { return nil, trace.BadParameter("duplicate key detected in PutBatch: %q", v) } // Many Backend implementations rely on unique keys for correct operation. - // Where it is up to the caller to ensure this to remove duplication keys - // Just in case we will fallback to single Put calls if duplicates are detected. + // Where it is up to the caller to ensure this to remove duplication keys. if v, ok := bk.(BatchPutter); ok { revs, err := v.PutBatch(ctx, items) return revs, trace.Wrap(err) @@ -412,6 +428,33 @@ func PutBatch(ctx context.Context, bk Backend, items []Item) ([]string, error) { return revisions, nil } +// DeleteBatch deletes multiple keys from the backend. If the backend implements +// [BatchDeleter], it delegates to the optimized batch implementation. Otherwise, +// it falls back to calling [Backend.Delete] for each key individually. +// Keys that do not exist are silently ignored. +// +// WARNING: Make sure that keys are unique when calling DeleteBatch. +// +// TODO(smallinsky): Move to backend interfaces and make required for backends to +// implement batch deletes interfaces. +func DeleteBatch(ctx context.Context, bk Backend, keys []Key) error { + keys = slices.DeduplicateKey(keys, func(k Key) string { return k.String() }) + + if v, ok := bk.(BatchDeleter); ok { + return trace.Wrap(v.DeleteBatch(ctx, keys)) + } + + for _, key := range keys { + if err := bk.Delete(ctx, key); err != nil { + if trace.IsNotFound(err) { + continue + } + return trace.Wrap(err) + } + } + return nil +} + func hasDuplicateKeys(items Items) (string, bool) { seen := make(map[string]struct{}) for _, ca := range items { diff --git a/lib/backend/report.go b/lib/backend/report.go index 525d79004f8..8316a8bdd25 100644 --- a/lib/backend/report.go +++ b/lib/backend/report.go @@ -81,6 +81,8 @@ func (r *ReporterConfig) CheckAndSetDefaults() error { } var _ Backend = (*Reporter)(nil) +var _ BatchDeleter = (*Reporter)(nil) +var _ BatchPutter = (*Reporter)(nil) // Reporter wraps a Backend implementation and reports // statistics about the backend operations @@ -545,6 +547,32 @@ func (s *Reporter) AtomicWrite(ctx context.Context, condacts []ConditionalAction return } +// DeleteBatch deletes multiple keys from the backend. +func (s *Reporter) DeleteBatch(ctx context.Context, keys []Key) error { + ctx, span := s.Tracer.Start( + ctx, + "backend/DeleteBatch", + oteltrace.WithAttributes( + attribute.Int("batch_size", len(keys)), + ), + ) + defer span.End() + + start := s.Clock().Now() + err := DeleteBatch(ctx, s.Backend, keys) + s.batchWriteLatencies.Observe(s.Clock().Since(start).Seconds()) + s.batchWriteRequests.Inc() + if err != nil { + s.batchWriteRequestsFailed.Inc() + } else { + s.writes.Add(float64(len(keys))) + } + for _, key := range keys { + s.trackRequest(ctx, types.OpDelete, key, Key{}) + } + return err +} + // DeleteRange deletes range of items func (s *Reporter) DeleteRange(ctx context.Context, startKey, endKey Key) error { ctx, span := s.Tracer.Start( diff --git a/lib/backend/sanitize.go b/lib/backend/sanitize.go index a0f0081ec83..402bdcc2ff5 100644 --- a/lib/backend/sanitize.go +++ b/lib/backend/sanitize.go @@ -75,6 +75,8 @@ func IsKeySafe(key Key) bool { } var _ Backend = (*Sanitizer)(nil) +var _ BatchDeleter = (*Sanitizer)(nil) +var _ BatchPutter = (*Sanitizer)(nil) // Sanitizer wraps a [Backend] implementation to make sure all // [Key]s written to the backend are allowed. Retrieval and deletion @@ -170,6 +172,11 @@ func (s *Sanitizer) Delete(ctx context.Context, key Key) error { return s.backend.Delete(ctx, key) } +// DeleteBatch deletes multiple items from the backend. +func (s *Sanitizer) DeleteBatch(ctx context.Context, keys []Key) error { + return trace.Wrap(DeleteBatch(ctx, s.backend, keys)) +} + // ConditionalDelete deletes the item by key if the revision matches the stored revision. func (s *Sanitizer) ConditionalDelete(ctx context.Context, key Key, revision string) error { return s.backend.ConditionalDelete(ctx, key, revision) diff --git a/lib/backend/test/delete_batch.go b/lib/backend/test/delete_batch.go new file mode 100644 index 00000000000..4e8c0059d7a --- /dev/null +++ b/lib/backend/test/delete_batch.go @@ -0,0 +1,188 @@ +/* + * Teleport + * Copyright (C) 2026 Gravitational, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +package test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/gravitational/teleport/api/types" + "github.com/gravitational/teleport/lib/backend" +) + +func runDeleteBatch(t *testing.T, newBackend Constructor) { + t.Helper() + + bk, _, err := newBackend() + require.NoError(t, err) + t.Cleanup(func() { _ = bk.Close() }) + + deleter, ok := bk.(backend.BatchDeleter) + if !ok { + t.Skip("backend does not implement DeleteBatch; skipping DeleteBatch suite") + } + + prefix := MakePrefix() + rangeStart := prefix("") + rangeEnd := backend.RangeEnd(prefix("")) + + putItems := func(t *testing.T, items []backend.Item) { + t.Helper() + for _, item := range items { + _, err := bk.Put(t.Context(), item) + require.NoError(t, err) + } + } + + newTestItems := func() []backend.Item { + return []backend.Item{ + {Key: prefix("a"), Value: []byte("A"), Expires: time.Now().Add(1 * time.Hour)}, + {Key: prefix("b"), Value: []byte("B")}, + {Key: prefix("c"), Value: []byte("C"), Expires: time.Now().Add(2 * time.Hour)}, + } + } + + t.Run("delete batch items should be propagated in event stream", func(t *testing.T) { + items := newTestItems() + putItems(t, items) + + w, err := bk.NewWatcher(t.Context(), backend.Watch{}) + require.NoError(t, err) + t.Cleanup(func() { w.Close() }) + + select { + case <-w.Done(): + t.Fatal("watcher closed immediately") + case ev := <-w.Events(): + require.Equal(t, types.OpInit, ev.Type) + } + + keys := make([]backend.Key, 0, len(items)) + for _, item := range items { + keys = append(keys, item.Key) + } + + err = deleter.DeleteBatch(t.Context(), keys) + require.NoError(t, err) + + got := waitForDeleteEvents(t, w, len(keys), watchEventTimeout) + require.Len(t, got, len(keys)) + for i, key := range keys { + require.Equal(t, 0, key.Compare(got[i])) + } + + // Confirm nothing remains. + res, err := bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit) + require.NoError(t, err) + require.Empty(t, res.Items) + }) + + t.Run("delete-then-verify-empty", func(t *testing.T) { + items := newTestItems() + putItems(t, items) + + // Verify items exist. + res, err := bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit) + require.NoError(t, err) + require.Len(t, res.Items, len(items)) + + keys := make([]backend.Key, 0, len(items)) + for _, item := range items { + keys = append(keys, item.Key) + } + + err = deleter.DeleteBatch(t.Context(), keys) + require.NoError(t, err) + + // Verify items are gone. + res, err = bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit) + require.NoError(t, err) + require.Empty(t, res.Items) + }) + + t.Run("delete-nonexistent-keys", func(t *testing.T) { + keys := []backend.Key{ + prefix("nonexistent1"), + prefix("nonexistent2"), + prefix("nonexistent3"), + } + + // Deleting keys that don't exist should not error. + err := deleter.DeleteBatch(t.Context(), keys) + require.NoError(t, err) + }) + + t.Run("delete-empty-batch", func(t *testing.T) { + err := deleter.DeleteBatch(t.Context(), nil) + require.NoError(t, err) + + err = deleter.DeleteBatch(t.Context(), []backend.Key{}) + require.NoError(t, err) + }) + + t.Run("delete-partial-existing", func(t *testing.T) { + items := newTestItems() + putItems(t, items) + + // Delete a mix of existing and non-existing keys. + keys := []backend.Key{ + items[0].Key, + prefix("nonexistent"), + items[2].Key, + } + + err := deleter.DeleteBatch(t.Context(), keys) + require.NoError(t, err) + + // Only the middle item should remain. + res, err := bk.GetRange(t.Context(), rangeStart, rangeEnd, backend.NoLimit) + require.NoError(t, err) + require.Len(t, res.Items, 1) + require.Equal(t, 0, items[1].Key.Compare(res.Items[0].Key)) + + require.NoError(t, bk.DeleteRange(t.Context(), rangeStart, rangeEnd)) + }) +} + +func waitForDeleteEvents(t *testing.T, w backend.Watcher, wantCount int, timeout time.Duration) []backend.Key { + t.Helper() + + var out []backend.Key + deadline := time.NewTimer(timeout) + defer deadline.Stop() + + for len(out) < wantCount { + select { + case ev, ok := <-w.Events(): + if !ok { + t.Fatalf("watcher closed before receiving all events: got=%d want=%d", len(out), wantCount) + } + if ev.Type == types.OpDelete { + out = append(out, ev.Item.Key) + } + case <-deadline.C: + t.Fatalf("timed out waiting for delete events: got=%d want=%d", len(out), wantCount) + case <-w.Done(): + t.Fatalf("watcher done before receiving all events: got=%d want=%d", len(out), wantCount) + } + } + return out +} diff --git a/lib/backend/test/suite.go b/lib/backend/test/suite.go index d59551fffa8..aae7c9fb0b4 100644 --- a/lib/backend/test/suite.go +++ b/lib/backend/test/suite.go @@ -202,6 +202,10 @@ func RunBackendComplianceSuite(t *testing.T, newBackend Constructor) { t.Run("PutBatch", func(t *testing.T) { runPutBatch(t, newBackend) }) + + t.Run("DeleteBatch", func(t *testing.T) { + runDeleteBatch(t, newBackend) + }) } // RequireItems asserts that the supplied `actual` items collection matches diff --git a/lib/backend/wrap.go b/lib/backend/wrap.go index 127c8002a43..3dc9752586a 100644 --- a/lib/backend/wrap.go +++ b/lib/backend/wrap.go @@ -29,6 +29,8 @@ import ( ) var _ Backend = (*Wrapper)(nil) +var _ BatchDeleter = (*Wrapper)(nil) +var _ BatchPutter = (*Wrapper)(nil) // Wrapper wraps a Backend implementation that can fail // on demand. @@ -123,6 +125,11 @@ func (s *Wrapper) Delete(ctx context.Context, key Key) error { return s.backend.Delete(ctx, key) } +// DeleteBatch deletes multiple items from the backend. +func (s *Wrapper) DeleteBatch(ctx context.Context, keys []Key) error { + return trace.Wrap(DeleteBatch(ctx, s.backend, keys)) +} + // ConditionalDelete deletes item by key if revisions match. func (s *Wrapper) ConditionalDelete(ctx context.Context, key Key, revision string) error { return s.backend.ConditionalDelete(ctx, key, revision)