mirror of
https://github.com/lukaszraczylo/traefikoidc.git
synced 2026-10-14 04:07:46 +00:00
R76: dcr singleflight, L1 no-resurrect, TLS loopback
This commit is contained in:
Vendored
+27
-9
@@ -561,9 +561,11 @@ func (h *HybridBackend) l1BackfillWorker() {
|
||||
for len(h.l1BackfillBuffer) > 0 {
|
||||
select {
|
||||
case item := <-h.l1BackfillBuffer:
|
||||
writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
_ = h.primary.Set(writeCtx, item.key, item.value, item.ttl)
|
||||
cancel()
|
||||
if h.keyStillInSecondary(item.key) {
|
||||
writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
_ = h.primary.Set(writeCtx, item.key, item.value, item.ttl)
|
||||
cancel()
|
||||
}
|
||||
default:
|
||||
return
|
||||
}
|
||||
@@ -574,17 +576,33 @@ func (h *HybridBackend) l1BackfillWorker() {
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
if err := h.primary.Set(writeCtx, item.key, item.value, item.ttl); err != nil {
|
||||
h.logger.Debugf("Failed to populate L1 cache from L2 for key %s: %v", redactKey(item.key), err)
|
||||
} else {
|
||||
h.logger.Debugf("Populated L1 cache from L2 for key: %s", redactKey(item.key))
|
||||
if h.keyStillInSecondary(item.key) {
|
||||
writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
if err := h.primary.Set(writeCtx, item.key, item.value, item.ttl); err != nil {
|
||||
h.logger.Debugf("Failed to populate L1 cache from L2 for key %s: %v", redactKey(item.key), err)
|
||||
} else {
|
||||
h.logger.Debugf("Populated L1 cache from L2 for key: %s", redactKey(item.key))
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// keyStillInSecondary reports whether the key still exists in the L2 cache.
|
||||
// Used by the L1 backfill worker: the value was captured from L2 at read
|
||||
// time, and a Delete may have run since. Writing it into L1 without this
|
||||
// check would resurrect a key the caller just removed from both tiers.
|
||||
func (h *HybridBackend) keyStillInSecondary(key string) bool {
|
||||
if h.fallbackMode.Load() {
|
||||
return true // No L2 available to check; keep behavior unchanged.
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
exists, err := h.secondary.Exists(ctx, key)
|
||||
return err != nil || exists
|
||||
}
|
||||
|
||||
// asyncWriteWorker processes asynchronous writes to L2
|
||||
func (h *HybridBackend) asyncWriteWorker() {
|
||||
defer h.wg.Done()
|
||||
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
//go:build !yaegi
|
||||
|
||||
package backends
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// gatedSetBackend wraps a mockBackend whose Set blocks until release is
|
||||
// closed (once). Lets the test park the backfill worker mid-flight.
|
||||
type gatedSetBackend struct {
|
||||
*mockBackend
|
||||
startOnce sync.Once
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func (b *gatedSetBackend) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error {
|
||||
b.startOnce.Do(func() { close(b.started) })
|
||||
<-b.release
|
||||
return b.mockBackend.Set(ctx, key, value, ttl)
|
||||
}
|
||||
|
||||
// TestHybridBackend_L1Backfill_NoResurrectAfterDelete regresses the L1
|
||||
// backfill worker resurrecting a key that was deleted after the L2 read
|
||||
// (Get) captured its value. The worker must re-check L2 and not write a
|
||||
// value into L1 for a key that is gone from L2.
|
||||
func TestHybridBackend_L1Backfill_NoResurrectAfterDelete(t *testing.T) {
|
||||
const key = "k"
|
||||
|
||||
inner := newMockBackend()
|
||||
gated := &gatedSetBackend{mockBackend: inner, started: make(chan struct{}), release: make(chan struct{})}
|
||||
secondary := newMockBackend()
|
||||
|
||||
hybrid, err := NewHybridBackend(&HybridConfig{
|
||||
Primary: gated,
|
||||
Secondary: secondary,
|
||||
AsyncBufferSize: 64,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
defer hybrid.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Park the worker on a filler backfill first so `key` (enqueued after
|
||||
// the filler) is processed only once we delete it.
|
||||
require.NoError(t, secondary.Set(ctx, "filler", []byte("x"), time.Minute))
|
||||
_, _, exists, err := hybrid.Get(ctx, "filler")
|
||||
require.NoError(t, err)
|
||||
require.True(t, exists)
|
||||
|
||||
select {
|
||||
case <-gated.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("worker never parked on filler Set")
|
||||
}
|
||||
|
||||
// Now trigger a backfill for `key` (this appends it behind the parked
|
||||
// filler), then delete it before the worker can reach it.
|
||||
require.NoError(t, secondary.Set(ctx, key, []byte("v"), time.Minute))
|
||||
_, _, exists, err = hybrid.Get(ctx, key)
|
||||
require.NoError(t, err)
|
||||
require.True(t, exists)
|
||||
_, err = hybrid.Delete(ctx, key)
|
||||
require.NoError(t, err)
|
||||
|
||||
close(gated.release)
|
||||
|
||||
// Wait for the worker to drain, failing if the deleted key ever appears
|
||||
// in L1 (resurrection). On the fixed code it re-checks L2 and skips;
|
||||
// on the buggy code it writes the stale value shortly after release.
|
||||
deadline := time.Now().Add(1 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if _, _, ok, _ := inner.Get(ctx, key); ok {
|
||||
t.Fatal("deleted key was resurrected into L1 by the backfill worker")
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user