Move unlock call to a background context in GC

This ensures that if the GC operation times out, the unlock still can run

internal/datastore/common/gc.go

@@ -196,7 +196,7 @@ func RunGarbageCollection(gc GarbageCollector, window, timeout time.Duration) er
 	}
 
 	defer func() {
-		err := gc.UnlockAfterGCRun(ctx)
+		err := gc.UnlockAfterGCRun(context.Background())
 		if err != nil {
 			log.Error().
 				Err(err).

internal/datastore/common/gc_test.go

@@ -23,6 +23,8 @@ type fakeGC struct {
 	deleter      gcDeleter
 	metrics      gcMetrics
 	lock         sync.RWMutex
+	wasLocked    bool
+	wasUnlocked  bool
 }
 
 type gcMetrics struct {
@@ -39,11 +41,16 @@ func newFakeGC(deleter gcDeleter) fakeGC {
 	}
 }
 
-func (*fakeGC) LockForGCRun(ctx context.Context) (bool, error) {
+func (f *fakeGC) LockForGCRun(ctx context.Context) (bool, error) {
+	f.wasLocked = true
 	return true, nil
 }
 
-func (*fakeGC) UnlockAfterGCRun(ctx context.Context) error {
+func (f *fakeGC) UnlockAfterGCRun(ctx context.Context) error {
+	if ctx.Err() != nil {
+		return ctx.Err()
+	}
+	f.wasUnlocked = true
 	return nil
 }
 
@@ -227,3 +234,23 @@ func TestGCFailureBackoffReset(t *testing.T) {
 	// the GC enough time to run.
 	require.Greater(t, gc.GetMetrics().markedCompleteCount, 20, "Next interval was not reset with backoff")
 }
+
+func TestGCUnlockOnTimeout(t *testing.T) {
+	gc := newFakeGC(alwaysErrorDeleter{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go func() {
+		interval := 10 * time.Millisecond
+		window := 10 * time.Second
+		timeout := 1 * time.Millisecond
+
+		require.Error(t, StartGarbageCollector(ctx, &gc, interval, window, timeout))
+	}()
+
+	time.Sleep(30 * time.Millisecond)
+	require.False(t, gc.HasGCRun(), "GC should not have run")
+	require.True(t, gc.wasLocked, "GC should have been locked")
+	require.True(t, gc.wasUnlocked, "GC should have been unlocked")
+}