From 667044b07b4e887d0c5dfb1f22f6efe222e21beb Mon Sep 17 00:00:00 2001 From: Dimitri Herzog Date: Sat, 28 Mar 2026 21:59:26 +0100 Subject: [PATCH] refactor: redis write through cache (#2025) --- .gitignore | 1 + cache/redis.go | 449 +++++++++++++++ cache/redis_suite_test.go | 18 + cache/redis_test.go | 733 ++++++++++++++++++++++++ e2e/redis_test.go | 224 ++++++++ evt/events.go | 17 + metrics/metrics_test.go | 1 + redis/event_bridge.go | 144 +++++ redis/event_bridge_test.go | 334 +++++++++++ redis/pubsub.go | 119 ++++ redis/redis.go | 325 +---------- redis/redis_suite_test.go | 7 - redis/redis_test.go | 355 ++---------- resolver/blocking_resolver.go | 72 ++- resolver/blocking_resolver_test.go | 86 +-- resolver/caching_resolver.go | 57 +- resolver/caching_resolver_test.go | 76 --- resolver/parallel_best_resolver_test.go | 22 +- resolver/resolver_test.go | 2 +- server/server.go | 79 ++- 20 files changed, 2243 insertions(+), 878 deletions(-) create mode 100644 cache/redis.go create mode 100644 cache/redis_suite_test.go create mode 100644 cache/redis_test.go create mode 100644 redis/event_bridge.go create mode 100644 redis/event_bridge_test.go create mode 100644 redis/pubsub.go diff --git a/.gitignore b/.gitignore index e4c451aa..4d7afbb0 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ bin/ dist/ docs/docs.go +docs/superpowers/ site/ config.yml todo.txt diff --git a/cache/redis.go b/cache/redis.go new file mode 100644 index 00000000..d4db91ab --- /dev/null +++ b/cache/redis.go @@ -0,0 +1,449 @@ +package cache + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/0xERR0R/blocky/log" + "github.com/0xERR0R/blocky/metrics" + blockredis "github.com/0xERR0R/blocky/redis" + goredis "github.com/go-redis/redis/v8" + "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/sirupsen/logrus" +) + +const ( + defaultBatchSize = 100 + defaultFlushInterval = 100 * time.Millisecond + defaultSendBufSize = 1000 + redisScanCount = 100 + drainTimeout = 5 * time.Second +) + +//nolint:gochecknoglobals +var redisBufferDrops = promauto.With(metrics.Reg).NewCounter( + prometheus.CounterOpts{ + Name: "blocky_redis_cache_buffer_drops_total", + Help: "Number of cache entries dropped because the Redis send buffer was full", + }, +) + +// RedisOptions configures the RedisExpiringCache decorator. +type RedisOptions[T any] struct { + // Prefix is prepended to all Redis keys. + Prefix string + // Channel is the pub/sub channel name used for cross-instance sync. + Channel string + // Encode serializes a cache value to bytes. + Encode func(*T) ([]byte, error) + // Decode deserializes bytes back to a cache value. + Decode func([]byte) (*T, error) + // BatchSize is the maximum number of entries per Redis pipeline flush. + BatchSize int + // FlushInterval is the maximum time between flushes. + FlushInterval time.Duration + // SendBufSize is the capacity of the internal send buffer channel. + SendBufSize int +} + +type redisSyncEntry struct { + Key string `json:"k"` + Data []byte `json:"d"` +} + +type redisSyncMessage struct { + Entries []redisSyncEntry `json:"e"` + Client string `json:"c"` +} + +type sendBufferEntry[T any] struct { + key string + val *T + expiration time.Duration +} + +// RedisExpiringCache wraps an ExpiringCache[T] with transparent Redis +// write-through and pub/sub synchronisation across instances. +type RedisExpiringCache[T any] struct { + inner ExpiringCache[T] + client *goredis.Client + opts RedisOptions[T] + instanceID string + sendBuf chan sendBufferEntry[T] + logger *logrus.Entry +} + +// NewRedisExpiringByteCache creates a RedisExpiringCache for []byte values, +// using identity encode/decode when no codec is provided. +func NewRedisExpiringByteCache( + ctx context.Context, + inner ExpiringCache[[]byte], + client *goredis.Client, + opts RedisOptions[[]byte], +) (*RedisExpiringCache[[]byte], error) { + if opts.Encode == nil && opts.Decode == nil { + opts.Encode = func(b *[]byte) ([]byte, error) { return *b, nil } + opts.Decode = func(b []byte) (*[]byte, error) { + cp := make([]byte, len(b)) + copy(cp, b) + + return &cp, nil + } + } + + return NewRedisExpiringCache(ctx, inner, client, opts) +} + +// NewRedisExpiringCache creates a new RedisExpiringCache decorator. +// +// It performs a blocking startup scan of existing Redis keys and loads them +// into inner before launching the background writer and subscriber goroutines. +// The goroutines run until ctx is cancelled. +func NewRedisExpiringCache[T any]( + ctx context.Context, + inner ExpiringCache[T], + client *goredis.Client, + opts RedisOptions[T], +) (*RedisExpiringCache[T], error) { + if opts.Encode == nil || opts.Decode == nil { + return nil, errors.New("RedisOptions: Encode and Decode must both be set") + } + + if opts.BatchSize <= 0 { + opts.BatchSize = defaultBatchSize + } + + if opts.FlushInterval <= 0 { + opts.FlushInterval = defaultFlushInterval + } + + if opts.SendBufSize <= 0 { + opts.SendBufSize = defaultSendBufSize + } + + c := &RedisExpiringCache[T]{ + inner: inner, + client: client, + opts: opts, + instanceID: uuid.NewString(), + sendBuf: make(chan sendBufferEntry[T], opts.SendBufSize), + logger: log.PrefixedLog("redis-cache"), + } + + // Blocking startup load. + if err := c.loadFromRedis(ctx); err != nil { + c.logger.WithError(err).Warn("startup Redis scan failed – starting with empty local cache") + } + + go c.runSubscriber(ctx) + go c.runWriter(ctx) + + return c, nil +} + +// Put stores the value in the inner cache immediately, then enqueues a +// non-blocking send to Redis. If the send buffer is full the entry is dropped +// and a warning is logged – the DNS path is never blocked. +func (c *RedisExpiringCache[T]) Put(key string, val *T, expiration time.Duration) { + c.inner.Put(key, val, expiration) + + if expiration <= 0 { + return + } + + entry := sendBufferEntry[T]{key: key, val: val, expiration: expiration} + + select { + case c.sendBuf <- entry: + default: + redisBufferDrops.Inc() + c.logger.Warn("redis send buffer full, dropping cache entry") + } +} + +// Get reads from the inner cache only. +func (c *RedisExpiringCache[T]) Get(key string) (*T, time.Duration) { + return c.inner.Get(key) +} + +// TotalCount delegates to the inner cache. +func (c *RedisExpiringCache[T]) TotalCount() int { + return c.inner.TotalCount() +} + +const clearTimeout = 10 * time.Second + +// Clear removes all entries from the inner cache and from Redis (all keys +// matching the configured prefix). +func (c *RedisExpiringCache[T]) Clear() { + c.inner.Clear() + + ctx, cancel := context.WithTimeout(context.Background(), clearTimeout) + defer cancel() + + err := c.scanKeys(ctx, func(ctx context.Context, keys []string) error { + return c.client.Del(ctx, keys...).Err() + }) + if err != nil { + c.logger.WithError(err).Warn("Redis Clear failed") + } +} + +// redisKey returns the prefixed Redis key for the given cache key. +func (c *RedisExpiringCache[T]) redisKey(key string) string { + return c.opts.Prefix + key +} + +// scanKeys iterates over all Redis keys matching the configured prefix +// and calls fn for each batch of keys returned by SCAN. +func (c *RedisExpiringCache[T]) scanKeys( + ctx context.Context, fn func(ctx context.Context, keys []string) error, +) error { + pattern := c.opts.Prefix + "*" + + var cursor uint64 + + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, pattern, redisScanCount).Result() + if err != nil { + return fmt.Errorf("SCAN: %w", err) + } + + if len(keys) > 0 { + if err := fn(ctx, keys); err != nil { + return err + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + return nil +} + +// loadFromRedis scans existing Redis keys and populates the inner cache. +// It uses a pipeline per SCAN page to minimise round trips. +func (c *RedisExpiringCache[T]) loadFromRedis(ctx context.Context) error { + return c.scanKeys(ctx, func(ctx context.Context, keys []string) error { + if err := c.loadKeysViaPipeline(ctx, keys); err != nil { + c.logger.WithError(err).Warn("pipeline load failed for SCAN page") + } + + return nil + }) +} + +// loadKeysViaPipeline fetches TTL and value for a batch of keys in a single pipeline round trip. +func (c *RedisExpiringCache[T]) loadKeysViaPipeline(ctx context.Context, keys []string) error { + pipe := c.client.Pipeline() + + ttlCmds := make([]*goredis.DurationCmd, len(keys)) + getCmds := make([]*goredis.StringCmd, len(keys)) + + for i, rk := range keys { + ttlCmds[i] = pipe.TTL(ctx, rk) + getCmds[i] = pipe.Get(ctx, rk) + } + + if _, err := pipe.Exec(ctx); err != nil && ctx.Err() == nil { + return fmt.Errorf("pipeline exec: %w", err) + } + + for i, rk := range keys { + dur, err := ttlCmds[i].Result() + if err != nil || dur <= 0 { + continue + } + + data, err := getCmds[i].Bytes() + if err != nil { + continue + } + + val, err := c.opts.Decode(data) + if err != nil { + c.logger.WithError(err).Warn("failed to decode Redis entry on startup") + + continue + } + + cacheKey := strings.TrimPrefix(rk, c.opts.Prefix) + c.inner.Put(cacheKey, val, dur) + } + + return nil +} + +// runWriter is the background goroutine that drains sendBuf and flushes +// batches to Redis via a pipeline + a single PUBLISH. +func (c *RedisExpiringCache[T]) runWriter(ctx context.Context) { + ticker := time.NewTicker(c.opts.FlushInterval) + defer ticker.Stop() + + batch := make([]sendBufferEntry[T], 0, c.opts.BatchSize) + + flush := func(fctx context.Context) { + if len(batch) == 0 { + return + } + + c.flushBatch(fctx, batch) + batch = make([]sendBufferEntry[T], 0, c.opts.BatchSize) + } + + for { + select { + case entry := <-c.sendBuf: + batch = append(batch, entry) + if len(batch) >= c.opts.BatchSize { + flush(ctx) + } + + case <-ticker.C: + flush(ctx) + + case <-ctx.Done(): + drainCtx, drainCancel := context.WithTimeout(context.Background(), drainTimeout) + + // Drain remaining buffer before exiting. + drainLoop: + for { + select { + case entry := <-c.sendBuf: + batch = append(batch, entry) + if len(batch) >= c.opts.BatchSize { + flush(drainCtx) //nolint:contextcheck + } + default: + break drainLoop + } + } + + flush(drainCtx) //nolint:contextcheck + drainCancel() + + return + } + } +} + +// flushBatch writes a slice of entries to Redis via a pipeline and publishes a +// single sync message on the configured channel. +func (c *RedisExpiringCache[T]) flushBatch(ctx context.Context, batch []sendBufferEntry[T]) { + syncEntries := make([]redisSyncEntry, 0, len(batch)) + + pipe := c.client.Pipeline() + + for _, e := range batch { + data, err := c.opts.Encode(e.val) + if err != nil { + c.logger.WithError(err).Warn("failed to encode cache entry for Redis") + + continue + } + + pipe.Set(ctx, c.redisKey(e.key), data, e.expiration) + + syncEntries = append(syncEntries, redisSyncEntry{ + Key: e.key, + Data: data, + }) + } + + if _, err := pipe.Exec(ctx); err != nil && ctx.Err() == nil { + c.logger.WithError(err).Warn("Redis pipeline exec failed") + } + + if len(syncEntries) == 0 || c.opts.Channel == "" { + return + } + + msg := redisSyncMessage{ + Entries: syncEntries, + Client: c.instanceID, + } + + payload, err := json.Marshal(msg) + if err != nil { + c.logger.WithError(err).Warn("failed to marshal sync message") + + return + } + + if err := c.client.Publish(ctx, c.opts.Channel, payload).Err(); err != nil && ctx.Err() == nil { + c.logger.WithError(err).Warn("Redis PUBLISH failed") + } +} + +// runSubscriber subscribes to the pub/sub channel and applies remote entries +// to the inner cache, filtering out messages originating from this instance. +// It reconnects automatically on channel closure with exponential backoff. +func (c *RedisExpiringCache[T]) runSubscriber(ctx context.Context) { + if c.opts.Channel == "" { + return + } + + loop := &blockredis.PubSubLoop{ + Client: c.client, + Channel: c.opts.Channel, + Logger: c.logger, + Handler: func(ctx context.Context, payload string) { + c.handleSyncMessage(ctx, []byte(payload)) + }, + } + + loop.Run(ctx) +} + +// handleSyncMessage decodes and applies a pub/sub sync message, skipping +// entries that originated from this instance. +func (c *RedisExpiringCache[T]) handleSyncMessage(ctx context.Context, payload []byte) { + var msg redisSyncMessage + if err := json.Unmarshal(payload, &msg); err != nil { + c.logger.WithError(err).Warn("failed to decode sync message") + + return + } + + if msg.Client == c.instanceID { + return + } + + pipe := c.client.Pipeline() + + ttlCmds := make([]*goredis.DurationCmd, len(msg.Entries)) + for i, e := range msg.Entries { + ttlCmds[i] = pipe.TTL(ctx, c.redisKey(e.Key)) + } + + if _, err := pipe.Exec(ctx); err != nil { + c.logger.WithError(err).Warn("failed to fetch TTLs for sync entries") + + return + } + + for i, e := range msg.Entries { + ttl, err := ttlCmds[i].Result() + if err != nil || ttl <= 0 { + continue + } + + val, err := c.opts.Decode(e.Data) + if err != nil { + c.logger.WithError(err).Warn("failed to decode sync entry") + + continue + } + + c.inner.Put(e.Key, val, ttl) + } +} diff --git a/cache/redis_suite_test.go b/cache/redis_suite_test.go new file mode 100644 index 00000000..b03dddf6 --- /dev/null +++ b/cache/redis_suite_test.go @@ -0,0 +1,18 @@ +package cache + +import ( + "testing" + + "github.com/0xERR0R/blocky/log" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func init() { + log.Silence() +} + +func TestCacheSuite(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Cache Suite") +} diff --git a/cache/redis_test.go b/cache/redis_test.go new file mode 100644 index 00000000..fb052af4 --- /dev/null +++ b/cache/redis_test.go @@ -0,0 +1,733 @@ +package cache + +import ( + "context" + "encoding/json" + "errors" + "time" + + expirationcache "github.com/0xERR0R/expiration-cache" + "github.com/alicebob/miniredis/v2" + goredis "github.com/go-redis/redis/v8" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// testValue is a simple string wrapper used as cache value type in tests. +type testValue struct { + Data string `json:"data"` +} + +func encodeTestValue(v *testValue) ([]byte, error) { + return json.Marshal(v) +} + +func decodeTestValue(b []byte) (*testValue, error) { + var v testValue + if err := json.Unmarshal(b, &v); err != nil { + return nil, err + } + + return &v, nil +} + +// newTestInner creates a fresh in-memory expiration cache. +func newTestInner(ctx context.Context) ExpiringCache[testValue] { + return expirationcache.NewCache[testValue](ctx, expirationcache.Options{}) +} + +// defaultOpts returns a RedisOptions with the given prefix suitable for tests. +func defaultOpts(prefix string) RedisOptions[testValue] { + return RedisOptions[testValue]{ + Prefix: prefix, + Channel: "test-sync-" + prefix, + Encode: encodeTestValue, + Decode: decodeTestValue, + BatchSize: 100, + FlushInterval: 20 * time.Millisecond, + SendBufSize: 1000, + } +} + +// newRedisClient creates a goredis client pointing at the given miniredis server. +func newRedisClient(srv *miniredis.Miniredis) *goredis.Client { + return goredis.NewClient(&goredis.Options{Addr: srv.Addr()}) +} + +var _ = Describe("RedisExpiringCache", func() { + var ( + srv *miniredis.Miniredis + client *goredis.Client + ) + + BeforeEach(func() { + var err error + srv, err = miniredis.Run() + Expect(err).ToNot(HaveOccurred()) + + DeferCleanup(srv.Close) + + client = newRedisClient(srv) + DeferCleanup(func() { _ = client.Close() }) + }) + + Describe("Put", func() { + When("a value is stored", func() { + It("stores in inner AND Redis", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("put:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("foo", &testValue{Data: "bar"}, time.Minute) + + // Inner cache must have the entry immediately. + val, _ := inner.Get("foo") + Expect(val).ToNot(BeNil()) + Expect(val.Data).To(Equal("bar")) + + // Redis key must appear after async flush. + Eventually(func() bool { + exists, _ := client.Exists(ctx, "put:foo").Result() + + return exists > 0 + }).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue()) + }) + }) + + When("the send buffer is full", func() { + It("completes without blocking", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("full:") + opts.SendBufSize = 2 + opts.FlushInterval = 10 * time.Minute // prevent flush during test + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + done := make(chan struct{}) + go func() { + defer close(done) + + for i := 0; i < 10; i++ { + c.Put("key", &testValue{Data: "v"}, time.Minute) + } + }() + + Eventually(done).WithTimeout(time.Second).Should(BeClosed()) + }) + }) + }) + + Describe("Get", func() { + When("a value exists only in Redis (not in inner cache)", func() { + It("returns nil", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + // Pre-populate Redis with a key under a different prefix so + // the decorator's startup scan does not pick it up. + data, _ := encodeTestValue(&testValue{Data: "direct"}) + Expect(client.Set(ctx, "other:foo", data, time.Minute).Err()).ToNot(HaveOccurred()) + + // Create decorator with a prefix that does NOT match the key above. + innerEmpty := newTestInner(ctx) + optsEmpty := defaultOpts("get-empty:") + + c, err := NewRedisExpiringCache(ctx, innerEmpty, client, optsEmpty) + Expect(err).ToNot(HaveOccurred()) + + val, _ := c.Get("foo") + Expect(val).To(BeNil()) + }) + }) + }) + + Describe("Clear", func() { + When("entries exist in both inner cache and Redis", func() { + It("clears both", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("clear:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("foo", &testValue{Data: "bar"}, time.Minute) + + // Wait for Redis write. + Eventually(func() bool { + exists, _ := client.Exists(ctx, "clear:foo").Result() + + return exists > 0 + }).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue()) + + c.Clear() + + Expect(inner.TotalCount()).To(Equal(0)) + + keys, err := client.Keys(ctx, "clear:*").Result() + Expect(err).ToNot(HaveOccurred()) + Expect(keys).To(BeEmpty()) + }) + }) + }) + + Describe("Cross-instance sync", func() { + When("two instances share the same Redis", func() { + It("propagates Put from instance1 to instance2's inner cache", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + sharedOpts := defaultOpts("sync:") + sharedOpts.FlushInterval = 10 * time.Millisecond + + inner1 := newTestInner(ctx) + c1, err := NewRedisExpiringCache(ctx, inner1, client, sharedOpts) + Expect(err).ToNot(HaveOccurred()) + + inner2 := newTestInner(ctx) + c2, err := NewRedisExpiringCache(ctx, inner2, client, sharedOpts) + Expect(err).ToNot(HaveOccurred()) + + _ = c2 // used via inner2 + + c1.Put("shared", &testValue{Data: "hello"}, time.Minute) + + Eventually(func() *testValue { + val, _ := inner2.Get("shared") + + return val + }).WithTimeout(3 * time.Second).WithPolling(20 * time.Millisecond).ShouldNot(BeNil()) + }) + }) + }) + + Describe("Echo filtering", func() { + When("an instance publishes a message", func() { + It("does not double-count the entry in its own cache", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("echo:") + opts.FlushInterval = 10 * time.Millisecond + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("key", &testValue{Data: "v"}, time.Minute) + + // Wait for the flush to happen. + Eventually(func() bool { + exists, _ := client.Exists(ctx, "echo:key").Result() + + return exists > 0 + }).WithTimeout(2 * time.Second).WithPolling(10 * time.Millisecond).Should(BeTrue()) + + // TotalCount must remain 1, not grow due to echo. + Consistently(func() int { + return c.TotalCount() + }, "300ms", "30ms").Should(Equal(1)) + }) + }) + }) + + Describe("Startup cache load", func() { + When("Redis already contains keys with the configured prefix", func() { + It("loads those entries into the inner cache on creation", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + data, _ := encodeTestValue(&testValue{Data: "pre-existing"}) + Expect(client.Set(ctx, "load:existing", data, time.Minute).Err()).ToNot(HaveOccurred()) + + inner := newTestInner(ctx) + opts := defaultOpts("load:") + + _, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + val, _ := inner.Get("existing") + Expect(val).ToNot(BeNil()) + Expect(val.Data).To(Equal("pre-existing")) + }) + }) + }) + + Describe("Put with zero expiration", func() { + When("expiration is zero", func() { + It("stores in inner but does not enqueue to Redis", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("zero-exp:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("no-redis", &testValue{Data: "val"}, 0) + + // Inner should not cache it either (expiration <= 0 means no cache) + val, _ := inner.Get("no-redis") + Expect(val).To(BeNil()) + + // Redis should never get this key + Consistently(func() bool { + exists, _ := client.Exists(ctx, "zero-exp:no-redis").Result() + + return exists > 0 + }, "200ms", "50ms").Should(BeFalse()) + }) + }) + }) + + Describe("TotalCount", func() { + When("multiple values are stored", func() { + It("returns the count from the inner cache", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("count:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("a", &testValue{Data: "1"}, time.Minute) + c.Put("b", &testValue{Data: "2"}, time.Minute) + c.Put("c", &testValue{Data: "3"}, time.Minute) + + Expect(c.TotalCount()).To(Equal(3)) + }) + }) + }) + + Describe("handleSyncMessage", func() { + When("the payload is invalid JSON", func() { + It("does not add entries to the cache", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("bad-json:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.handleSyncMessage(ctx, []byte("not-valid-json{{{")) + + Expect(inner.TotalCount()).To(Equal(0)) + }) + }) + + When("the message is from the same instance", func() { + It("skips the message", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("self:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + data, _ := encodeTestValue(&testValue{Data: "val"}) + msg := redisSyncMessage{ + Entries: []redisSyncEntry{{Key: "k", Data: data}}, + Client: c.instanceID, + } + payload, _ := json.Marshal(msg) + + c.handleSyncMessage(ctx, payload) + + Expect(inner.TotalCount()).To(Equal(0)) + }) + }) + + When("a sync entry has invalid data", func() { + It("skips the bad entry but processes valid ones", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("bad-entry:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + goodData, _ := encodeTestValue(&testValue{Data: "good"}) + + // Store keys in Redis so handleSyncMessage can fetch their TTLs. + client.Set(ctx, "bad-entry:bad", []byte("not-json"), time.Minute) + client.Set(ctx, "bad-entry:good", goodData, time.Minute) + + msg := redisSyncMessage{ + Entries: []redisSyncEntry{ + {Key: "bad", Data: []byte("not-json")}, + {Key: "good", Data: goodData}, + }, + Client: "other-instance", + } + payload, _ := json.Marshal(msg) + + c.handleSyncMessage(ctx, payload) + + Expect(inner.TotalCount()).To(Equal(1)) + val, _ := inner.Get("good") + Expect(val).ToNot(BeNil()) + Expect(val.Data).To(Equal("good")) + }) + }) + }) + + Describe("flushBatch", func() { + When("encoding fails for an entry", func() { + It("skips the bad entry and flushes the rest", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("enc-err:") + + callCount := 0 + opts.Encode = func(v *testValue) ([]byte, error) { + callCount++ + if v.Data == "fail" { + return nil, errors.New("encode error") + } + + return encodeTestValue(v) + } + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + batch := []sendBufferEntry[testValue]{ + {key: "fail-key", val: &testValue{Data: "fail"}, expiration: time.Minute}, + {key: "ok-key", val: &testValue{Data: "ok"}, expiration: time.Minute}, + } + c.flushBatch(ctx, batch) + + // Only the "ok" entry should be in Redis + exists, _ := client.Exists(ctx, "enc-err:ok-key").Result() + Expect(exists).To(Equal(int64(1))) + + exists, _ = client.Exists(ctx, "enc-err:fail-key").Result() + Expect(exists).To(Equal(int64(0))) + }) + }) + + When("the channel is empty", func() { + It("writes to Redis but does not publish sync messages", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("no-chan:") + opts.Channel = "" + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + batch := []sendBufferEntry[testValue]{ + {key: "k", val: &testValue{Data: "v"}, expiration: time.Minute}, + } + c.flushBatch(ctx, batch) + + // Entry should still be in Redis + exists, _ := client.Exists(ctx, "no-chan:k").Result() + Expect(exists).To(Equal(int64(1))) + }) + }) + }) + + Describe("NewRedisExpiringCache defaults", func() { + When("options have zero values for batch size, flush interval, and send buf size", func() { + It("applies defaults", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := RedisOptions[testValue]{ + Prefix: "defaults:", + Channel: "defaults-chan", + Encode: encodeTestValue, + Decode: decodeTestValue, + // All zero — should get defaults + } + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + Expect(c.opts.BatchSize).To(Equal(100)) + Expect(c.opts.FlushInterval).To(Equal(100 * time.Millisecond)) + Expect(c.opts.SendBufSize).To(Equal(1000)) + }) + }) + }) + + Describe("Clear with scan error", func() { + When("the Redis connection fails during Clear", func() { + It("does not panic and clears the inner cache", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("clear-err:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("foo", &testValue{Data: "bar"}, time.Minute) + + // Close the Redis server to cause scan failure + srv.Close() + + c.Clear() + + // Inner cache should still be cleared + Expect(inner.TotalCount()).To(Equal(0)) + }) + }) + }) + + Describe("loadKeysViaPipeline with decode errors", func() { + When("a stored value cannot be decoded", func() { + It("skips that key and continues", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + // Store a key with bad data directly in Redis + Expect(client.Set(ctx, "decode-err:bad", "not-valid-json", time.Minute).Err()).ToNot(HaveOccurred()) + + // Store a key with good data + goodData, _ := encodeTestValue(&testValue{Data: "good"}) + Expect(client.Set(ctx, "decode-err:good", goodData, time.Minute).Err()).ToNot(HaveOccurred()) + + inner := newTestInner(ctx) + opts := defaultOpts("decode-err:") + + _, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + // Only the good key should be loaded + val, _ := inner.Get("good") + Expect(val).ToNot(BeNil()) + Expect(val.Data).To(Equal("good")) + + // The bad key should be skipped + badVal, _ := inner.Get("bad") + Expect(badVal).To(BeNil()) + }) + }) + }) + + Describe("runWriter batch size trigger", func() { + When("enough entries are buffered to fill a batch", func() { + It("flushes immediately without waiting for the timer", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("batch:") + opts.BatchSize = 3 + opts.FlushInterval = 10 * time.Minute // prevent timer flush + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + // Put exactly BatchSize entries to trigger immediate flush + c.Put("a", &testValue{Data: "1"}, time.Minute) + c.Put("b", &testValue{Data: "2"}, time.Minute) + c.Put("c", &testValue{Data: "3"}, time.Minute) + + Eventually(func() bool { + exists, _ := client.Exists(ctx, "batch:a", "batch:b", "batch:c").Result() + + return exists == 3 + }).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue()) + }) + }) + }) + + Describe("runSubscriber with empty channel", func() { + When("the Channel option is empty", func() { + It("does not subscribe to pub/sub", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("no-sub:") + opts.Channel = "" + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + // Put a value via the cache (will write to Redis but not publish sync) + c.Put("local", &testValue{Data: "local-only"}, time.Minute) + + val, _ := c.Get("local") + Expect(val).ToNot(BeNil()) + Expect(val.Data).To(Equal("local-only")) + }) + }) + }) + + Describe("Graceful degradation", func() { + When("the Redis client points to a dead server", func() { + It("Put and Get still work locally", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + deadClient := goredis.NewClient(&goredis.Options{ + Addr: "127.0.0.1:0", + DialTimeout: 50 * time.Millisecond, + ReadTimeout: 50 * time.Millisecond, + }) + DeferCleanup(deadClient.Close) + + inner := newTestInner(ctx) + opts := defaultOpts("dead:") + + // Startup scan will fail but should not return error (just log). + c, err := NewRedisExpiringCache(ctx, inner, deadClient, opts) + Expect(err).ToNot(HaveOccurred()) + + c.Put("local", &testValue{Data: "value"}, time.Minute) + + val, _ := c.Get("local") + Expect(val).ToNot(BeNil()) + Expect(val.Data).To(Equal("value")) + }) + }) + }) + + Describe("handleSyncMessage with TTL fetch failure", func() { + When("the Redis server is down during TTL fetch", func() { + It("skips all entries without panic", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("ttl-fail:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + goodData, _ := encodeTestValue(&testValue{Data: "val"}) + + // Close Redis to cause TTL fetch failure + srv.Close() + + msg := redisSyncMessage{ + Entries: []redisSyncEntry{{Key: "k", Data: goodData}}, + Client: "other-instance", + } + payload, _ := json.Marshal(msg) + + c.handleSyncMessage(ctx, payload) + + Expect(inner.TotalCount()).To(Equal(0)) + }) + }) + }) + + Describe("handleSyncMessage with missing TTL", func() { + When("a sync entry's key has no expiration in Redis", func() { + It("skips the entry", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("no-ttl:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + data, _ := encodeTestValue(&testValue{Data: "val"}) + + // Store key WITHOUT expiration (TTL = -1 in Redis) + client.Set(ctx, "no-ttl:k", data, 0) + + msg := redisSyncMessage{ + Entries: []redisSyncEntry{{Key: "k", Data: data}}, + Client: "other-instance", + } + payload, _ := json.Marshal(msg) + + c.handleSyncMessage(ctx, payload) + + Expect(inner.TotalCount()).To(Equal(0)) + }) + }) + }) + + Describe("flushBatch with publish failure", func() { + When("the Redis server goes down after entries are written", func() { + It("does not panic", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := newTestInner(ctx) + opts := defaultOpts("pub-fail:") + + c, err := NewRedisExpiringCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + // Close Redis so both pipeline exec and publish will fail + srv.Close() + + batch := []sendBufferEntry[testValue]{ + {key: "k", val: &testValue{Data: "v"}, expiration: time.Minute}, + } + Expect(func() { c.flushBatch(ctx, batch) }).ToNot(Panic()) + }) + }) + }) + + Describe("NewRedisExpiringByteCache", func() { + When("no codec is provided", func() { + It("uses identity encode/decode functions", func() { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + + inner := expirationcache.NewCache[[]byte](ctx, expirationcache.Options{}) + + opts := RedisOptions[[]byte]{ + Prefix: "bytes:", + Channel: "bytes-chan", + // Encode and Decode intentionally nil — defaults should apply + } + + c, err := NewRedisExpiringByteCache(ctx, inner, client, opts) + Expect(err).ToNot(HaveOccurred()) + + original := []byte("hello world") + c.Put("test", &original, time.Minute) + + // Wait for Redis write + Eventually(func() bool { + exists, _ := client.Exists(ctx, "bytes:test").Result() + + return exists > 0 + }).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue()) + + // Verify round-trip: read from Redis and decode + data, err := client.Get(ctx, "bytes:test").Bytes() + Expect(err).ToNot(HaveOccurred()) + Expect(data).To(Equal(original)) + }) + }) + }) +}) diff --git a/e2e/redis_test.go b/e2e/redis_test.go index 557d70bb..53d907a5 100644 --- a/e2e/redis_test.go +++ b/e2e/redis_test.go @@ -2,6 +2,8 @@ package e2e import ( "context" + "net" + "net/http" "strings" . "github.com/0xERR0R/blocky/helpertest" @@ -168,6 +170,228 @@ var _ = Describe("Redis configuration tests", func() { }) }) }) + + Describe("Blocking state sync via Redis", func() { + When("Redis, blocking, and 2 blocky instances are configured", func() { + BeforeEach(func(ctx context.Context) { + _, err = createDNSMokkaContainer(ctx, "moka2", e2eNet, + `A blocked.com/NOERROR("A 5.6.7.8 123")`, + ) + Expect(err).Should(Succeed()) + + _, err = createHTTPServerContainer(ctx, "httpserver", e2eNet, "list.txt", "blocked.com") + Expect(err).Should(Succeed()) + + blocky1, err = createBlockyContainerFromString(ctx, e2eNet, dedent(` + log: + level: warn + upstreams: + groups: + default: + - moka2 + ports: + http: 4000 + blocking: + denylists: + ads: + - http://httpserver:8080/list.txt + clientGroupsBlock: + default: + - ads + redis: + address: redis:6379 + `)) + Expect(err).Should(Succeed()) + + blocky2, err = createBlockyContainerFromString(ctx, e2eNet, dedent(` + log: + level: warn + upstreams: + groups: + default: + - moka2 + ports: + http: 4000 + blocking: + denylists: + ads: + - http://httpserver:8080/list.txt + clientGroupsBlock: + default: + - ads + redis: + address: redis:6379 + `)) + Expect(err).Should(Succeed()) + }) + + It("syncs blocking disable/enable between instances", func(ctx context.Context) { + msg := util.NewMsgWithQuestion("blocked.com.", A) + + By("verifying blocking works on both instances", func() { + Eventually(doDNSRequest, "5s", "2ms").WithArguments(ctx, blocky1, msg). + Should(BeDNSRecord("blocked.com.", A, "0.0.0.0")) + Eventually(doDNSRequest, "5s", "2ms").WithArguments(ctx, blocky2, msg). + Should(BeDNSRecord("blocked.com.", A, "0.0.0.0")) + }) + + By("disabling blocking on instance1 via API", func() { + host, port, err := getContainerHostPort(ctx, blocky1, "4000/tcp") + Expect(err).Should(Succeed()) + + resp, err := http.Get("http://" + net.JoinHostPort(host, port) + "/api/blocking/disable") + Expect(err).Should(Succeed()) + defer resp.Body.Close() + Expect(resp.StatusCode).Should(Equal(http.StatusOK)) + }) + + By("verifying instance2 also has blocking disabled", func() { + Eventually(doDNSRequest, "5s", "100ms").WithArguments(ctx, blocky2, msg). + Should(BeDNSRecord("blocked.com.", A, "5.6.7.8")) + }) + + By("re-enabling blocking on instance1", func() { + host, port, err := getContainerHostPort(ctx, blocky1, "4000/tcp") + Expect(err).Should(Succeed()) + + resp, err := http.Get("http://" + net.JoinHostPort(host, port) + "/api/blocking/enable") + Expect(err).Should(Succeed()) + defer resp.Body.Close() + Expect(resp.StatusCode).Should(Equal(http.StatusOK)) + }) + + By("verifying instance2 re-enables blocking", func() { + Eventually(doDNSRequest, "5s", "100ms").WithArguments(ctx, blocky2, msg). + Should(BeDNSRecord("blocked.com.", A, "0.0.0.0")) + }) + }) + }) + }) + + Describe("Cache clear propagation via Redis", func() { + When("Redis and 2 blocky instances share cached entries", func() { + BeforeEach(func(ctx context.Context) { + blocky1, err = createBlockyContainerFromString(ctx, e2eNet, dedent(` + log: + level: warn + upstreams: + groups: + default: + - moka1 + ports: + http: 4000 + caching: + minTime: 5m + redis: + address: redis:6379 + `)) + Expect(err).Should(Succeed()) + }) + + It("clears cache entries from Redis when flushed via API", func(ctx context.Context) { + msg := util.NewMsgWithQuestion("google.de.", A) + + By("populating cache via instance1", func() { + Eventually(doDNSRequest, "5s", "2ms").WithArguments(ctx, blocky1, msg). + Should(BeDNSRecord("google.de.", A, "1.2.3.4")) + }) + + By("waiting for Redis to have the entry", func() { + Eventually(dbSize, "5s", "2ms").WithArguments(ctx, redisClient). + Should(BeNumerically(">=", 1)) + }) + + By("flushing cache via instance1 API", func() { + host, port, err := getContainerHostPort(ctx, blocky1, "4000/tcp") + Expect(err).Should(Succeed()) + + resp, err := http.Post( + "http://"+net.JoinHostPort(host, port)+"/api/cache/flush", + "application/json", nil) + Expect(err).Should(Succeed()) + defer resp.Body.Close() + Expect(resp.StatusCode).Should(Equal(http.StatusOK)) + }) + + By("verifying Redis cache was cleared", func() { + Eventually(dbSize, "5s", "100ms").WithArguments(ctx, redisClient). + Should(BeNumerically("==", 0)) + }) + }) + }) + }) + + Describe("Multiple query types via Redis cache", func() { + When("both A and AAAA queries are cached", func() { + BeforeEach(func(ctx context.Context) { + _, err = createDNSMokkaContainer(ctx, "moka2", e2eNet, + `A google/NOERROR("A 1.2.3.4 123")`, + `AAAA google/NOERROR("AAAA 2001:db8::1 123")`, + ) + Expect(err).Should(Succeed()) + + blocky1, err = createBlockyContainerFromString(ctx, e2eNet, dedent(` + log: + level: warn + upstreams: + groups: + default: + - moka2 + redis: + address: redis:6379 + `)) + Expect(err).Should(Succeed()) + + blocky2, err = createBlockyContainerFromString(ctx, e2eNet, dedent(` + log: + level: warn + upstreams: + groups: + default: + - moka2 + redis: + address: redis:6379 + `)) + Expect(err).Should(Succeed()) + }) + + It("shares both A and AAAA cached responses via Redis", func(ctx context.Context) { + By("querying A and AAAA on instance1", func() { + Eventually(doDNSRequest, "5s", "2ms"). + WithArguments(ctx, blocky1, util.NewMsgWithQuestion("google.de.", A)). + Should(BeDNSRecord("google.de.", A, "1.2.3.4")) + + Eventually(doDNSRequest, "5s", "2ms"). + WithArguments(ctx, blocky1, util.NewMsgWithQuestion("google.de.", AAAA)). + Should(BeDNSRecord("google.de.", AAAA, "2001:db8::1")) + }) + + By("waiting for Redis entries", func() { + Eventually(dbSize, "5s", "2ms").WithArguments(ctx, redisClient). + Should(BeNumerically(">=", 2)) + }) + + By("shutting down upstream", func() { + Expect(mokka.Terminate(ctx)).Should(Succeed()) + }) + + By("querying instance2 for both types from cache", func() { + Eventually(doDNSRequest, "5s", "2ms"). + WithArguments(ctx, blocky2, util.NewMsgWithQuestion("google.de.", A)). + Should(BeDNSRecord("google.de.", A, "1.2.3.4")) + + Eventually(doDNSRequest, "5s", "2ms"). + WithArguments(ctx, blocky2, util.NewMsgWithQuestion("google.de.", AAAA)). + Should(BeDNSRecord("google.de.", AAAA, "2001:db8::1")) + }) + + By("No warnings/errors in log", func() { + Expect(getContainerLogs(ctx, blocky1)).Should(BeEmpty()) + Expect(getContainerLogs(ctx, blocky2)).Should(BeEmpty()) + }) + }) + }) + }) }) func dbSize(ctx context.Context, redisClient *redis.Client) (int64, error) { diff --git a/evt/events.go b/evt/events.go index 17d6caec..12152164 100644 --- a/evt/events.go +++ b/evt/events.go @@ -1,6 +1,8 @@ package evt import ( + "time" + "github.com/asaskevich/EventBus" ) @@ -8,6 +10,14 @@ const ( // BlockingEnabledEvent fires if blocking status will be changed. Parameter: boolean (enabled = true) BlockingEnabledEvent = "blocking:enabled" + // BlockingStateChanged fires when blocking state changes locally (for Redis bridge). + // Parameter: BlockingState + BlockingStateChanged = "blocking:stateChanged" + + // BlockingStateChangedRemote fires when blocking state changes from a remote instance via Redis. + // Parameter: BlockingState + BlockingStateChangedRemote = "blocking:stateChangedRemote" + // BlockingCacheGroupChanged fires, if a list group is changed. Parameter: list type, group name, element count BlockingCacheGroupChanged = "blocking:cachingGroupChanged" @@ -37,3 +47,10 @@ var evtBus = EventBus.New() func Bus() EventBus.Bus { return evtBus } + +// BlockingState carries the full blocking state for cross-instance sync. +type BlockingState struct { + Enabled bool `json:"enabled"` + Duration time.Duration `json:"duration,omitempty"` + Groups []string `json:"groups,omitempty"` +} diff --git a/metrics/metrics_test.go b/metrics/metrics_test.go index 2ac2ee29..09587397 100644 --- a/metrics/metrics_test.go +++ b/metrics/metrics_test.go @@ -61,6 +61,7 @@ func AssertRegistryComplete(t *testing.T, reg *prometheus.Registry) { "blocky_cache_entries", "blocky_cache_hits_total", "blocky_cache_misses_total", + "blocky_redis_cache_buffer_drops_total", "blocky_last_list_group_refresh_timestamp_seconds", "blocky_prefetches_total", "blocky_prefetch_hits_total", diff --git a/redis/event_bridge.go b/redis/event_bridge.go new file mode 100644 index 00000000..83033da2 --- /dev/null +++ b/redis/event_bridge.go @@ -0,0 +1,144 @@ +package redis + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/0xERR0R/blocky/evt" + "github.com/0xERR0R/blocky/log" + goredis "github.com/go-redis/redis/v8" + "github.com/google/uuid" + "github.com/sirupsen/logrus" +) + +const ( + EventBridgeChannel = "blocky_sync_enabled" + bridgePublishTimeout = 5 * time.Second +) + +type bridgeMessage struct { + State evt.BlockingState `json:"s"` + Client string `json:"c"` +} + +// EventBusBridge connects the local event bus with Redis pub/sub for blocking state synchronization. +type EventBusBridge struct { + client *goredis.Client + id string + channel string + l *logrus.Entry + cancel context.CancelFunc + done <-chan struct{} + once sync.Once +} + +// NewEventBusBridge creates a new EventBusBridge that synchronizes blocking state +// between local event bus and Redis pub/sub. +func NewEventBusBridge(ctx context.Context, client *goredis.Client) (*EventBusBridge, error) { + ctx, cancel := context.WithCancel(ctx) + + b := &EventBusBridge{ + client: client, + id: uuid.NewString(), + channel: EventBridgeChannel, + l: log.PrefixedLog("redis-event-bridge"), + cancel: cancel, + done: ctx.Done(), + } + + if err := evt.Bus().Subscribe(evt.BlockingStateChanged, b.onLocalStateChanged); err != nil { + cancel() + + return nil, err + } + + ps := client.Subscribe(ctx, b.channel) + + if _, err := ps.Receive(ctx); err != nil { + _ = ps.Close() + _ = evt.Bus().Unsubscribe(evt.BlockingStateChanged, b.onLocalStateChanged) + cancel() + + return nil, err + } + + loop := &PubSubLoop{ + Client: client, + Channel: b.channel, + Logger: b.l, + Handler: b.handleMessage, + } + + go func() { + defer b.Close() + + loop.RunWithSub(ctx, ps) + }() + + return b, nil +} + +// Close stops the subscriber goroutine and unsubscribes from the local event bus. +// It is safe to call multiple times. +func (b *EventBusBridge) Close() error { + var unsubErr error + + b.once.Do(func() { + b.cancel() + unsubErr = evt.Bus().Unsubscribe(evt.BlockingStateChanged, b.onLocalStateChanged) + }) + + return unsubErr +} + +// onLocalStateChanged is called when a local blocking state change occurs. +// It marshals the state and publishes it to Redis. +func (b *EventBusBridge) onLocalStateChanged(state evt.BlockingState) { + msg := bridgeMessage{ + State: state, + Client: b.id, + } + + payload, err := json.Marshal(msg) + if err != nil { + b.l.Error("failed to marshal bridge message: ", err) + + return + } + + pubCtx, cancel := context.WithTimeout(context.Background(), bridgePublishTimeout) + defer cancel() + + select { + case <-b.done: + return + default: + } + + if err := b.client.Publish(pubCtx, b.channel, payload).Err(); err != nil { + b.l.Error("failed to publish to Redis: ", err) + } +} + +// handleMessage processes a single Redis pub/sub message, publishing remote +// state changes to the local event bus and filtering out echoes. +func (b *EventBusBridge) handleMessage(_ context.Context, payload string) { + if len(payload) == 0 { + return + } + + var bm bridgeMessage + if err := json.Unmarshal([]byte(payload), &bm); err != nil { + b.l.Error("failed to unmarshal bridge message: ", err) + + return + } + + if bm.Client == b.id { + return + } + + evt.Bus().Publish(evt.BlockingStateChangedRemote, bm.State) +} diff --git a/redis/event_bridge_test.go b/redis/event_bridge_test.go new file mode 100644 index 00000000..c717b2cb --- /dev/null +++ b/redis/event_bridge_test.go @@ -0,0 +1,334 @@ +package redis + +import ( + "context" + "encoding/json" + "time" + + "github.com/0xERR0R/blocky/evt" + "github.com/alicebob/miniredis/v2" + goredis "github.com/go-redis/redis/v8" + "github.com/google/uuid" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("EventBusBridge", func() { + var ( + redisServer *miniredis.Miniredis + redisClient *goredis.Client + bridge *EventBusBridge + ctx context.Context + cancel context.CancelFunc + ) + + BeforeEach(func() { + var err error + redisServer, err = miniredis.Run() + Expect(err).Should(Succeed()) + DeferCleanup(redisServer.Close) + + redisClient = goredis.NewClient(&goredis.Options{ + Addr: redisServer.Addr(), + }) + DeferCleanup(func() { _ = redisClient.Close() }) + + ctx, cancel = context.WithCancel(context.Background()) + DeferCleanup(cancel) + + bridge, err = NewEventBusBridge(ctx, redisClient) + Expect(err).Should(Succeed()) + DeferCleanup(func() { bridge.Close() }) + }) + + Describe("NewEventBusBridge errors", func() { + When("Redis is not reachable", func() { + It("should return an error", func() { + deadClient := goredis.NewClient(&goredis.Options{ + Addr: "127.0.0.1:0", + DialTimeout: 50 * time.Millisecond, + }) + DeferCleanup(func() { _ = deadClient.Close() }) + + newCtx, newCancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + DeferCleanup(newCancel) + + _, err := NewEventBusBridge(newCtx, deadClient) + Expect(err).Should(HaveOccurred()) + }) + }) + }) + + Describe("Local event → Redis publish", func() { + When("BlockingStateChanged is published on the local bus", func() { + It("should publish a message to the Redis channel that a second subscriber receives", func(specCtx context.Context) { + // Create a second bridge to act as receiver + bridge2, err := NewEventBusBridge(ctx, redisClient) + Expect(err).Should(Succeed()) + DeferCleanup(func() { bridge2.Close() }) + + receivedStates := make(chan evt.BlockingState, 1) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + expectedState := evt.BlockingState{Enabled: true} + evt.Bus().Publish(evt.BlockingStateChanged, expectedState) + + Eventually(receivedStates).WithTimeout(2 * time.Second).WithPolling(50 * time.Millisecond).Should(Receive(Equal(expectedState))) + }, SpecTimeout(3*time.Second)) + }) + }) + + Describe("Redis message → local event", func() { + When("a bridgeMessage with a different UUID arrives on Redis", func() { + It("should fire BlockingStateChangedRemote on the local bus with the correct state", func(specCtx context.Context) { + receivedStates := make(chan evt.BlockingState, 1) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + otherID := uuid.NewString() + + expectedState := evt.BlockingState{ + Enabled: true, + Duration: 5 * time.Minute, + Groups: []string{"ads", "malware"}, + } + + payload, err := json.Marshal(bridgeMessage{ + State: expectedState, + Client: otherID, + }) + Expect(err).Should(Succeed()) + + redisServer.Publish(EventBridgeChannel, string(payload)) + + Eventually(receivedStates).Should(Receive(Equal(expectedState))) + }, SpecTimeout(3*time.Second)) + }) + }) + + Describe("Echo filtering", func() { + When("a bridgeMessage with the bridge's own UUID arrives on Redis", func() { + It("should NOT fire BlockingStateChangedRemote on the local bus", func(specCtx context.Context) { + receivedStates := make(chan evt.BlockingState, 1) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + payload, err := json.Marshal(bridgeMessage{ + State: evt.BlockingState{Enabled: true}, + Client: bridge.id, + }) + Expect(err).Should(Succeed()) + + redisServer.Publish(EventBridgeChannel, string(payload)) + + Consistently(receivedStates, "300ms", "50ms").Should(BeEmpty()) + }, SpecTimeout(3*time.Second)) + }) + }) + + Describe("Close", func() { + When("Close is called multiple times", func() { + It("should be idempotent and not panic", func() { + Expect(bridge.Close()).Should(Succeed()) + Expect(bridge.Close()).Should(Succeed()) + }) + }) + }) + + Describe("consumeMessages", func() { + When("a message with empty payload arrives", func() { + It("should ignore it without error", func(specCtx context.Context) { + receivedStates := make(chan evt.BlockingState, 1) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + // Publish empty payload + redisServer.Publish(EventBridgeChannel, "") + + Consistently(receivedStates, "300ms", "50ms").Should(BeEmpty()) + }, SpecTimeout(3*time.Second)) + }) + + When("a message with invalid JSON arrives", func() { + It("should skip it without firing an event", func(specCtx context.Context) { + receivedStates := make(chan evt.BlockingState, 1) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + redisServer.Publish(EventBridgeChannel, "not-valid-json{{{") + + Consistently(receivedStates, "300ms", "50ms").Should(BeEmpty()) + }, SpecTimeout(3*time.Second)) + }) + }) + + Describe("onLocalStateChanged after close", func() { + When("the bridge context is cancelled before publishing", func() { + It("should not attempt to publish to Redis", func() { + // Cancel the bridge context + cancel() + + // onLocalStateChanged should return early due to done channel + // This should not panic or block + bridge.onLocalStateChanged(evt.BlockingState{Enabled: true}) + }) + }) + }) + + Describe("subscribeLoop reconnection", func() { + When("the Redis pub/sub connection closes unexpectedly", func() { + It("should not panic when pub/sub connection drops", func(specCtx context.Context) { + receivedStates := make(chan evt.BlockingState, 5) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + otherID := uuid.NewString() + + // Send a message before restart to verify connectivity + preState := evt.BlockingState{Enabled: true} + payload, err := json.Marshal(bridgeMessage{State: preState, Client: otherID}) + Expect(err).Should(Succeed()) + + redisServer.Publish(EventBridgeChannel, string(payload)) + + Eventually(receivedStates, "2s", "50ms").Should(Receive(Equal(preState))) + + // Restart miniredis to simulate connection drop + recovery + redisServer.Close() + redisServer2, err := miniredis.Run() + Expect(err).Should(Succeed()) + DeferCleanup(redisServer2.Close) + + // Note: since miniredis restarts on a new port, the existing bridge + // won't reconnect to it. But we verify the reconnect loop runs + // without panicking by cancelling context shortly after. + time.Sleep(100 * time.Millisecond) + }, SpecTimeout(5*time.Second)) + }) + + When("the context is cancelled during reconnect", func() { + It("should stop and return without error", func(specCtx context.Context) { + // Create a new bridge with its own context + bridgeCtx, bridgeCancel := context.WithCancel(context.Background()) + + b, err := NewEventBusBridge(bridgeCtx, redisClient) + Expect(err).Should(Succeed()) + + // Close the Redis server to trigger reconnection + redisServer.Close() + + // Give the bridge time to notice the disconnection + time.Sleep(100 * time.Millisecond) + + // Cancel the context to stop reconnection + bridgeCancel() + + // Verify the bridge shuts down cleanly + Eventually(func() error { + return b.Close() + }).WithTimeout(2 * time.Second).Should(Succeed()) + }, SpecTimeout(5*time.Second)) + }) + }) + + Describe("onLocalStateChanged error handling", func() { + When("Redis is unreachable during publish", func() { + It("does not panic", func() { + // Close Redis to cause publish failure + redisServer.Close() + + Expect(func() { + bridge.onLocalStateChanged(evt.BlockingState{Enabled: true}) + }).ToNot(Panic()) + }) + }) + }) + + Describe("Full payload round-trip", func() { + When("BlockingStateChanged is published locally with Duration and Groups", func() { + It("should preserve Duration and Groups through local → Redis → remote", func(specCtx context.Context) { + receivedStates := make(chan evt.BlockingState, 1) + + handler := func(state evt.BlockingState) { + receivedStates <- state + } + + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + DeferCleanup(func() { + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + }) + + // Create a second bridge acting as the "remote" receiver + bridge2, err := NewEventBusBridge(ctx, redisClient) + Expect(err).Should(Succeed()) + DeferCleanup(func() { bridge2.Close() }) + + // Override bridge2's handler to forward to our channel + Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed()) + + // Publish from bridge (bridge2 should receive since UUIDs differ) + originalState := evt.BlockingState{ + Enabled: false, + Duration: 2*time.Hour + 30*time.Minute, + Groups: []string{"group1", "group2", "group3"}, + } + + // Directly publish to Redis as bridge (simulate local→Redis path) + payload, err := json.Marshal(bridgeMessage{ + State: originalState, + Client: bridge.id, // from bridge, NOT bridge2 + }) + Expect(err).Should(Succeed()) + + redisServer.Publish(EventBridgeChannel, string(payload)) + + // bridge2 should receive it (different UUID) + Eventually(receivedStates, "2s", "50ms").Should(Receive(Equal(originalState))) + }, SpecTimeout(5*time.Second)) + }) + }) +}) diff --git a/redis/pubsub.go b/redis/pubsub.go new file mode 100644 index 00000000..2455d9a0 --- /dev/null +++ b/redis/pubsub.go @@ -0,0 +1,119 @@ +package redis + +import ( + "context" + "time" + + goredis "github.com/go-redis/redis/v8" + "github.com/sirupsen/logrus" +) + +const ( + reconnectBaseDelay = 500 * time.Millisecond + reconnectMaxDelay = 30 * time.Second +) + +// PubSubLoop manages a Redis pub/sub subscription with automatic reconnection. +// It calls handler for each received message payload and reconnects with +// exponential backoff when the channel closes unexpectedly. +type PubSubLoop struct { + Client *goredis.Client + Channel string + Logger *logrus.Entry + Handler func(ctx context.Context, payload string) +} + +// Run subscribes to the channel and processes messages until ctx is cancelled. +// It reconnects automatically on unexpected channel closure. +func (p *PubSubLoop) Run(ctx context.Context) { + p.RunWithSub(ctx, nil) +} + +// RunWithSub is like Run but uses an existing subscription for the first iteration. +// If initial is nil, a new subscription is created. +func (p *PubSubLoop) RunWithSub(ctx context.Context, initial *goredis.PubSub) { + sub := initial + if sub == nil { + sub = p.Client.Subscribe(ctx, p.Channel) + + if _, err := sub.Receive(ctx); err != nil { + p.Logger.WithError(err).Warn("Redis pub/sub initial subscribe failed, attempting to reconnect") + _ = sub.Close() + + sub = p.reconnect(ctx) + if sub == nil { + return + } + } + } + + for { + p.consume(ctx, sub) + _ = sub.Close() + + if ctx.Err() != nil { + return + } + + p.Logger.Warn("Redis pub/sub channel closed unexpectedly, attempting to reconnect") + + sub = p.reconnect(ctx) + if sub == nil { + return + } + } +} + +func (p *PubSubLoop) consume(ctx context.Context, sub *goredis.PubSub) { + ch := sub.Channel() + + for { + select { + case msg, ok := <-ch: + if !ok { + return + } + + if msg == nil { + continue + } + + p.Handler(ctx, msg.Payload) + + case <-ctx.Done(): + return + } + } +} + +// reconnect attempts to re-subscribe with exponential backoff. +// Returns nil if the context is cancelled before reconnecting. +func (p *PubSubLoop) reconnect(ctx context.Context) *goredis.PubSub { + delay := reconnectBaseDelay + + for { + select { + case <-ctx.Done(): + return nil + case <-time.After(delay): + } + + sub := p.Client.Subscribe(ctx, p.Channel) + + if _, err := sub.Receive(ctx); err != nil { + _ = sub.Close() + p.Logger.WithError(err).Warn("Redis pub/sub reconnect failed, retrying") + + delay *= 2 + if delay > reconnectMaxDelay { + delay = reconnectMaxDelay + } + + continue + } + + p.Logger.Info("Redis pub/sub reconnected successfully") + + return sub + } +} diff --git a/redis/redis.go b/redis/redis.go index 19c221c3..2ea28a0d 100644 --- a/redis/redis.go +++ b/redis/redis.go @@ -1,81 +1,22 @@ package redis import ( - "bytes" "context" - "encoding/json" "fmt" - "math" - "strings" - "time" "github.com/0xERR0R/blocky/config" - "github.com/0xERR0R/blocky/log" - "github.com/0xERR0R/blocky/model" - "github.com/0xERR0R/blocky/util" - "github.com/go-redis/redis/v8" - "github.com/google/uuid" - "github.com/miekg/dns" - "github.com/sirupsen/logrus" + goredis "github.com/go-redis/redis/v8" ) -const ( - SyncChannelName = "blocky_sync" - CacheStorePrefix = "blocky:cache:" - chanCap = 1000 - cacheReason = "EXTERNAL_CACHE" - defaultCacheTime = 1 * time.Second - messageTypeCache = 0 - messageTypeEnable = 1 -) - -// sendBuffer message -type bufferMessage struct { - Key string - Message *dns.Msg -} - -// redis pubsub message -type redisMessage struct { - Key string `json:"k,omitempty"` - Type int `json:"t"` - Message []byte `json:"m"` - Client []byte `json:"c"` -} - -// CacheChannel message -type CacheMessage struct { - Key string - Response *model.Response -} - -type EnabledMessage struct { - State bool `json:"s"` - Duration time.Duration `json:"d,omitempty"` - Groups []string `json:"g,omitempty"` -} - -// Client for redis communication -type Client struct { - config *config.Redis - client *redis.Client - l *logrus.Entry - id []byte - sendBuffer chan *bufferMessage - CacheChannel chan *CacheMessage - EnabledChannel chan *EnabledMessage -} - -// New creates a new redis client -func New(ctx context.Context, cfg *config.Redis) (*Client, error) { - // disable redis if no address is provided +// New creates a new Redis connection. Returns nil if Redis is not configured. +func New(ctx context.Context, cfg *config.Redis) (*goredis.Client, error) { if cfg == nil || len(cfg.Address) == 0 { return nil, nil //nolint:nilnil } - var baseClient *redis.Client + var client *goredis.Client if len(cfg.SentinelAddresses) > 0 { - baseClient = redis.NewFailoverClient(&redis.FailoverOptions{ + client = goredis.NewFailoverClient(&goredis.FailoverOptions{ MasterName: cfg.Address, SentinelUsername: cfg.Username, SentinelPassword: cfg.SentinelPassword, @@ -87,7 +28,7 @@ func New(ctx context.Context, cfg *config.Redis) (*Client, error) { MaxRetryBackoff: cfg.ConnectionCooldown.ToDuration(), }) } else { - baseClient = redis.NewClient(&redis.Options{ + client = goredis.NewClient(&goredis.Options{ Addr: cfg.Address, Username: cfg.Username, Password: cfg.Password, @@ -97,257 +38,11 @@ func New(ctx context.Context, cfg *config.Redis) (*Client, error) { }) } - rdb := baseClient.WithContext(ctx) + rdb := client.WithContext(ctx) - _, err := rdb.Ping(ctx).Result() - if err == nil { - var id []byte - - id, err = uuid.New().MarshalBinary() - if err == nil { - // construct client - res := &Client{ - config: cfg, - client: rdb, - l: log.PrefixedLog("redis"), - id: id, - sendBuffer: make(chan *bufferMessage, chanCap), - CacheChannel: make(chan *CacheMessage, chanCap), - EnabledChannel: make(chan *EnabledMessage, chanCap), - } - - // start channel handling go routine - err = res.startup(ctx) - - return res, err - } + if _, err := rdb.Ping(ctx).Result(); err != nil { + return nil, fmt.Errorf("failed to connect to Redis at '%s': %w", cfg.Address, err) } - return nil, err -} - -// PublishCache publish cache to redis async -func (c *Client) PublishCache(key string, message *dns.Msg) { - if len(key) > 0 && message != nil { - c.sendBuffer <- &bufferMessage{ - Key: key, - Message: message, - } - } -} - -func (c *Client) PublishEnabled(ctx context.Context, state *EnabledMessage) { - binState, sErr := json.Marshal(state) - if sErr == nil { - binMsg, mErr := json.Marshal(redisMessage{ - Type: messageTypeEnable, - Message: binState, - Client: c.id, - }) - - if mErr == nil { - c.client.Publish(ctx, SyncChannelName, binMsg) - } - } -} - -// GetRedisCache reads the redis cache and publish it to the channel -func (c *Client) GetRedisCache(ctx context.Context) { - c.l.Debug("GetRedisCache") - - go func() { - iter := c.client.Scan(ctx, 0, prefixKey("*"), 0).Iterator() - if err := iter.Err(); err != nil { - c.l.Error("GetRedisCache ", err) - - return - } - - for iter.Next(ctx) { - response, err := c.getResponse(ctx, iter.Val()) - if err == nil { - if response != nil { - if !util.CtxSend(ctx, c.CacheChannel, response) { - return - } - } - } else { - c.l.Error("GetRedisCache ", err) - } - } - }() -} - -// startup starts a new goroutine for subscription and translation -func (c *Client) startup(ctx context.Context) error { - ps := c.client.Subscribe(ctx, SyncChannelName) - - _, err := ps.Receive(ctx) - if err == nil { - go func() { - for { - select { - // received message from subscription - case msg := <-ps.Channel(): - c.l.Debug("Received message: ", msg) - - if msg != nil && len(msg.Payload) > 0 { - // message is not empty - c.processReceivedMessage(ctx, msg) - } - // publish message from buffer - case s := <-c.sendBuffer: - c.publishMessageFromBuffer(ctx, s) - // context is done - case <-ctx.Done(): - c.client.Close() - - return - } - } - }() - } else { - return fmt.Errorf("failed to receive from Redis subscription channel '%s': %w", SyncChannelName, err) - } - - return nil -} - -func (c *Client) publishMessageFromBuffer(ctx context.Context, s *bufferMessage) { - origRes := s.Message - origRes.Compress = true - binRes, pErr := origRes.Pack() - - if pErr == nil { - binMsg, mErr := json.Marshal(redisMessage{ - Key: s.Key, - Type: messageTypeCache, - Message: binRes, - Client: c.id, - }) - - if mErr == nil { - c.client.Publish(ctx, SyncChannelName, binMsg) - } - - c.client.Set(ctx, - prefixKey(s.Key), - binRes, - c.getTTL(origRes)) - } -} - -func (c *Client) processReceivedMessage(ctx context.Context, msg *redis.Message) { - var rm redisMessage - - if err := json.Unmarshal([]byte(msg.Payload), &rm); err != nil { - c.l.Error("Processing error: ", err) - - return - } - - // message was sent from a different blocky instance - if !bytes.Equal(rm.Client, c.id) { - switch rm.Type { - case messageTypeCache: - var cm *CacheMessage - - cm, err := convertMessage(&rm, 0) - if err != nil { - c.l.Error("Processing CacheMessage error: ", err) - - return - } - - util.CtxSend(ctx, c.CacheChannel, cm) - case messageTypeEnable: - var msg EnabledMessage - - if err := json.Unmarshal(rm.Message, &msg); err != nil { - c.l.Error("Processing EnabledMessage error: ", err) - - return - } - - util.CtxSend(ctx, c.EnabledChannel, &msg) - default: - c.l.Warn("Unknown message type: ", rm.Type) - } - } -} - -// getResponse returns model.Response for a key -func (c *Client) getResponse(ctx context.Context, key string) (*CacheMessage, error) { - resp, err := c.client.Get(ctx, key).Result() - if err == nil { - var ttl time.Duration - ttl, err = c.client.TTL(ctx, key).Result() - - if err == nil { - var result *CacheMessage - - result, err = convertMessage(&redisMessage{ - Key: cleanKey(key), - Message: []byte(resp), - }, ttl) - if err != nil { - return nil, fmt.Errorf("conversion error: %w", err) - } - - return result, nil - } - } - - return nil, err -} - -// convertMessage converts redisMessage to CacheMessage -func convertMessage(message *redisMessage, ttl time.Duration) (*CacheMessage, error) { - msg := dns.Msg{} - - err := msg.Unpack(message.Message) - if err == nil { - if ttl > 0 { - for _, a := range msg.Answer { - a.Header().Ttl = uint32(ttl.Seconds()) - } - } - - res := &CacheMessage{ - Key: message.Key, - Response: &model.Response{ - RType: model.ResponseTypeCACHED, - Reason: cacheReason, - Res: &msg, - }, - } - - return res, nil - } - - return nil, fmt.Errorf("failed to unpack DNS message from Redis for key '%s': %w", message.Key, err) -} - -// getTTL of dns message or return defaultCacheTime if 0 -func (c *Client) getTTL(dns *dns.Msg) time.Duration { - ttl := uint32(math.MaxInt32) - for _, a := range dns.Answer { - ttl = min(ttl, a.Header().Ttl) - } - - if ttl == 0 { - return defaultCacheTime - } - - return time.Duration(ttl) * time.Second -} - -// prefixKey with CacheStorePrefix -func prefixKey(key string) string { - return fmt.Sprintf("%s%s", CacheStorePrefix, key) -} - -// cleanKey trims CacheStorePrefix prefix -func cleanKey(key string) string { - return strings.TrimPrefix(key, CacheStorePrefix) + return rdb, nil } diff --git a/redis/redis_suite_test.go b/redis/redis_suite_test.go index 525921c3..7a9a1fdf 100644 --- a/redis/redis_suite_test.go +++ b/redis/redis_suite_test.go @@ -1,25 +1,18 @@ package redis import ( - "context" "testing" "github.com/0xERR0R/blocky/log" - "github.com/go-redis/redis/v8" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) func init() { log.Silence() - redis.SetLogger(NoLogs{}) } func TestRedisClient(t *testing.T) { RegisterFailHandler(Fail) RunSpecs(t, "Redis Suite") } - -type NoLogs struct{} - -func (l NoLogs) Printf(context.Context, string, ...any) {} diff --git a/redis/redis_test.go b/redis/redis_test.go index 6f566633..d94afe1b 100644 --- a/redis/redis_test.go +++ b/redis/redis_test.go @@ -2,31 +2,16 @@ package redis import ( "context" - "encoding/json" - "time" "github.com/0xERR0R/blocky/config" - "github.com/0xERR0R/blocky/util" "github.com/alicebob/miniredis/v2" "github.com/creasty/defaults" - "github.com/google/uuid" - "github.com/miekg/dns" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) -const ( - exampleComKey = CacheStorePrefix + "example.com" -) - -var _ = Describe("Redis client", func() { - var ( - redisConfig *config.Redis - - redisClient *Client - - err error - ) +var _ = Describe("Redis connection factory", func() { + var redisConfig *config.Redis BeforeEach(func() { var rcfg config.Redis @@ -34,323 +19,65 @@ var _ = Describe("Redis client", func() { redisConfig = &rcfg }) - Describe("Client creation", func() { - When("redis configuration has no address", func() { - It("should return nil without error", func(ctx context.Context) { - Expect(New(ctx, redisConfig)).Should(BeNil()) - }) - }) - - When("redis configuration has invalid address", func() { - BeforeEach(func() { - redisConfig.Address = "127.0.0.1:0" - }) - - It("should fail with error", func(ctx context.Context) { - _, err = New(ctx, redisConfig) - Expect(err).Should(HaveOccurred()) - }) - }) - - When("sentinel is enabled without servers", func() { - BeforeEach(func() { - redisConfig.Address = "test" - redisConfig.SentinelAddresses = []string{"127.0.0.1:0"} - }) - - It("should fail with error", func(ctx context.Context) { - _, err = New(ctx, redisConfig) - Expect(err).Should(HaveOccurred()) - }) - }) - - When("redis configuration has invalid password", func() { - BeforeEach(func() { - setupRedisServer(redisConfig) - redisConfig.Password = "wrong" - }) - - It("should fail with error", func(ctx context.Context) { - _, err = New(ctx, redisConfig) - Expect(err).Should(HaveOccurred()) - }) + When("configuration has no address", func() { + It("should return nil without error", func(ctx context.Context) { + client, err := New(ctx, redisConfig) + Expect(err).Should(Succeed()) + Expect(client).Should(BeNil()) }) }) - Describe("Publish message", func() { - var redisServer *miniredis.Miniredis + When("configuration has invalid address", func() { BeforeEach(func() { - redisServer = setupRedisServer(redisConfig) + redisConfig.Address = "127.0.0.1:0" }) - When("Redis client publishes 'cache' message", func() { - It("One new entry with TTL > 0 should be persisted in the database", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - - By("Database is empty", func() { - Eventually(func() []string { - return redisServer.DB(redisConfig.Database).Keys() - }).Should(BeEmpty()) - }) - - By("publish new message with TTL > 0", func() { - res, err := util.NewMsgWithAnswer("example.com.", 123, dns.Type(dns.TypeA), "123.124.122.123") - - Expect(err).Should(Succeed()) - - redisClient.PublishCache("example.com", res) - }) - - By("Database has one entry with correct TTL", func() { - Eventually(func() bool { - return redisServer.DB(redisConfig.Database).Exists(exampleComKey) - }).Should(BeTrue()) - - ttl := redisServer.DB(redisConfig.Database).TTL(exampleComKey) - Expect(ttl.Seconds()).Should(BeNumerically("~", 123)) - }) - }) - - It("One new entry with default TTL should be persisted in the database", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - - By("Database is empty", func() { - Eventually(func() []string { - return redisServer.DB(redisConfig.Database).Keys() - }).Should(BeEmpty()) - }) - - By("publish new message with TTL = 0", func() { - res, err := util.NewMsgWithAnswer("example.com.", 0, dns.Type(dns.TypeA), "123.124.122.123") - - Expect(err).Should(Succeed()) - - redisClient.PublishCache("example.com", res) - }) - - By("Database has one entry with default TTL", func() { - Eventually(func() bool { - return redisServer.DB(redisConfig.Database).Exists(exampleComKey) - }).Should(BeTrue()) - - ttl := redisServer.DB(redisConfig.Database).TTL(exampleComKey) - Expect(ttl.Seconds()).Should(BeNumerically("~", defaultCacheTime.Seconds())) - }) - }) - }) - When("Redis client publishes 'enabled' message", func() { - It("should propagate the message over redis", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - - redisClient.PublishEnabled(ctx, &EnabledMessage{ - State: true, - }) - Eventually(func() map[string]int { - return redisServer.PubSubNumSub(SyncChannelName) - }).Should(HaveLen(1)) - }, SpecTimeout(time.Second*3)) + It("should fail with error", func(ctx context.Context) { + _, err := New(ctx, redisConfig) + Expect(err).Should(HaveOccurred()) }) }) - Describe("Receive message", func() { - var redisServer *miniredis.Miniredis + When("sentinel is enabled without servers", func() { BeforeEach(func() { - redisServer = setupRedisServer(redisConfig) + redisConfig.Address = "test" + redisConfig.SentinelAddresses = []string{"127.0.0.1:0"} }) - When("'enabled' message is received", func() { - It("should propagate the message over the channel", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - var binState []byte - binState, err = json.Marshal(EnabledMessage{State: true}) - Expect(err).Should(Succeed()) - - var id []byte - id, err = uuid.New().MarshalBinary() - Expect(err).Should(Succeed()) - - var binMsg []byte - binMsg, err = json.Marshal(redisMessage{ - Type: messageTypeEnable, - Message: binState, - Client: id, - }) - Expect(err).Should(Succeed()) - - lenE := len(redisClient.EnabledChannel) - - rec := redisServer.Publish(SyncChannelName, string(binMsg)) - Expect(rec).Should(Equal(1)) - - Eventually(func() chan *EnabledMessage { - return redisClient.EnabledChannel - }).Should(HaveLen(lenE + 1)) - }) - }) - When("'cache' message is received", func() { - It("should propagate the message over the channel", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - - res, err := util.NewMsgWithAnswer("example.com.", 123, dns.Type(dns.TypeA), "123.124.122.123") - - Expect(err).Should(Succeed()) - - var binState []byte - binState, err = res.Pack() - Expect(err).Should(Succeed()) - - var id []byte - id, err = uuid.New().MarshalBinary() - Expect(err).Should(Succeed()) - - var binMsg []byte - binMsg, err = json.Marshal(redisMessage{ - Key: "example.com", - Type: messageTypeCache, - Message: binState, - Client: id, - }) - Expect(err).Should(Succeed()) - - lenE := len(redisClient.CacheChannel) - - rec := redisServer.Publish(SyncChannelName, string(binMsg)) - Expect(rec).Should(Equal(1)) - - Eventually(func() chan *CacheMessage { - return redisClient.CacheChannel - }).Should(HaveLen(lenE + 1)) - }, SpecTimeout(time.Second*3)) - }) - When("wrong data is received", func() { - It("should not propagate the message over the channel if data is wrong", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - - var id []byte - id, err = uuid.New().MarshalBinary() - Expect(err).Should(Succeed()) - - var binMsg []byte - binMsg, err = json.Marshal(redisMessage{ - Key: "unknown", - Type: messageTypeCache, - Message: []byte("test"), - Client: id, - }) - Expect(err).Should(Succeed()) - - lenE := len(redisClient.EnabledChannel) - lenC := len(redisClient.CacheChannel) - - rec := redisServer.Publish(SyncChannelName, string(binMsg)) - Expect(rec).Should(Equal(1)) - - Eventually(func() chan *EnabledMessage { - return redisClient.EnabledChannel - }).Should(HaveLen(lenE)) - - Eventually(func() chan *CacheMessage { - return redisClient.CacheChannel - }).Should(HaveLen(lenC)) - }, SpecTimeout(time.Second*3)) - It("should not propagate the message over the channel if type is wrong", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - - var id []byte - id, err = uuid.New().MarshalBinary() - Expect(err).Should(Succeed()) - - var binMsg []byte - binMsg, err = json.Marshal(redisMessage{ - Key: "unknown", - Type: 99, - Message: []byte("test"), - Client: id, - }) - Expect(err).Should(Succeed()) - - lenE := len(redisClient.EnabledChannel) - lenC := len(redisClient.CacheChannel) - - rec := redisServer.Publish(SyncChannelName, string(binMsg)) - Expect(rec).Should(Equal(1)) - - // Use Consistently to verify channels don't receive invalid messages - Consistently(func() int { - return len(redisClient.EnabledChannel) - }, "200ms", "50ms").Should(Equal(lenE)) - - Consistently(func() int { - return len(redisClient.CacheChannel) - }, "200ms", "50ms").Should(Equal(lenC)) - }, SpecTimeout(time.Second*3)) + It("should fail with error", func(ctx context.Context) { + _, err := New(ctx, redisConfig) + Expect(err).Should(HaveOccurred()) }) }) - Describe("Read the redis cache and publish it to the channel", func() { - var redisServer *miniredis.Miniredis + When("configuration has valid address", func() { BeforeEach(func() { - redisServer = setupRedisServer(redisConfig) + redisServer, err := miniredis.Run() + Expect(err).Should(Succeed()) + DeferCleanup(redisServer.Close) + redisConfig.Address = redisServer.Addr() }) - When("GetRedisCache is called with valid database entries", func() { - It("Should read data from Redis and propagate it via cache channel", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) - By("Database is empty", func() { - Eventually(func() []string { - return redisServer.DB(redisConfig.Database).Keys() - }).Should(BeEmpty()) - }) - - By("Put valid data in Redis by publishing the cache entry", func() { - var res *dns.Msg - - res, err = util.NewMsgWithAnswer("example.com.", 123, dns.Type(dns.TypeA), "123.124.122.123") - - Expect(err).Should(Succeed()) - - redisClient.PublishCache("example.com", res) - }) - - By("Database has one entry now", func() { - Eventually(func() []string { - return redisServer.DB(redisConfig.Database).Keys() - }).Should(HaveLen(1)) - }) - - By("call GetRedisCache - It should read one entry from redis and propagate it via channel", func() { - redisClient.GetRedisCache(ctx) - - Eventually(redisClient.CacheChannel).Should(HaveLen(1)) - }) - }, SpecTimeout(time.Second*4)) + It("should return a connected client", func(ctx context.Context) { + client, err := New(ctx, redisConfig) + Expect(err).Should(Succeed()) + Expect(client).ShouldNot(BeNil()) }) - When("GetRedisCache is called and database contains not valid entry", func() { - It("Should do nothing (only log error)", func(ctx context.Context) { - redisClient, err = New(ctx, redisConfig) - Expect(err).Should(Succeed()) + }) - Expect(redisServer.DB(redisConfig.Database).Set(CacheStorePrefix+"test", "test")).Should(Succeed()) - redisClient.GetRedisCache(ctx) - Consistently(redisClient.CacheChannel).Should(BeEmpty()) - }, SpecTimeout(time.Second*2)) + When("configuration has invalid password", func() { + BeforeEach(func() { + redisServer, err := miniredis.Run() + Expect(err).Should(Succeed()) + DeferCleanup(redisServer.Close) + redisServer.RequireAuth("correct-password") + redisConfig.Address = redisServer.Addr() + redisConfig.Password = "wrong" + }) + + It("should fail with error", func(ctx context.Context) { + _, err := New(ctx, redisConfig) + Expect(err).Should(HaveOccurred()) }) }) }) - -func setupRedisServer(cfg *config.Redis) *miniredis.Miniredis { - redisServer, err := miniredis.Run() - Expect(err).Should(Succeed()) - DeferCleanup(redisServer.Close) - cfg.Address = redisServer.Addr() - - return redisServer -} diff --git a/resolver/blocking_resolver.go b/resolver/blocking_resolver.go index 11025487..e205972a 100644 --- a/resolver/blocking_resolver.go +++ b/resolver/blocking_resolver.go @@ -24,7 +24,6 @@ import ( "github.com/0xERR0R/blocky/lists" "github.com/0xERR0R/blocky/log" "github.com/0xERR0R/blocky/model" - "github.com/0xERR0R/blocky/redis" "github.com/0xERR0R/blocky/util" "github.com/miekg/dns" @@ -94,7 +93,6 @@ type BlockingResolver struct { allowlistOnlyGroups map[string]bool status *status clientGroupsBlock map[string][]string - redisClient *redis.Client fqdnIPCache cache.ExpiringCache[[]net.IP] } @@ -118,7 +116,6 @@ func clientGroupsBlock(cfg config.Blocking) map[string][]string { // NewBlockingResolver returns a new configured instance of the resolver func NewBlockingResolver(ctx context.Context, cfg config.Blocking, - redis *redis.Client, bootstrap *Bootstrap, ) (r *BlockingResolver, err error) { blockHandler, err := createBlockHandler(cfg) @@ -152,7 +149,6 @@ func NewBlockingResolver(ctx context.Context, enableTimer: time.NewTimer(0), }, clientGroupsBlock: clientGroupsBlock(cfg), - redisClient: redis, } res.fqdnIPCache = expirationcache.NewCacheWithOnExpired[[]net.IP](ctx, expirationcache.Options{ @@ -161,43 +157,44 @@ func NewBlockingResolver(ctx context.Context, return res.queryForFQIdentifierIPs(ctx, key) }) - if res.redisClient != nil { - go res.redisSubscriber(ctx) - } - - err = evt.Bus().SubscribeOnce(evt.ApplicationStarted, func(_ ...string) { - go res.initFQDNIPCache(ctx) - }) - if err != nil { - return nil, fmt.Errorf("failed to subscribe to ApplicationStarted event: %w", err) + if err := res.subscribeEvents(ctx); err != nil { + return nil, err } return res, nil } -func (r *BlockingResolver) redisSubscriber(ctx context.Context) { - ctx, logger := r.log(ctx) +func (r *BlockingResolver) subscribeEvents(ctx context.Context) error { + err := evt.Bus().SubscribeOnce(evt.ApplicationStarted, func(_ ...string) { + go r.initFQDNIPCache(ctx) + }) + if err != nil { + return fmt.Errorf("failed to subscribe to ApplicationStarted event: %w", err) + } - for { - select { - case em := <-r.redisClient.EnabledChannel: - if em != nil { - logger.Debug("Received state from redis: ", em) - - if em.State { - r.internalEnableBlocking() - } else { - err := r.internalDisableBlocking(ctx, em.Duration, em.Groups) - if err != nil { - logger.Warn("Blocking couldn't be disabled:", err) - } + remoteHandler := func(state evt.BlockingState) { + go func() { + if state.Enabled { + r.internalEnableBlocking() + } else { + if disableErr := r.internalDisableBlocking(ctx, state.Duration, state.Groups); disableErr != nil { + log.PrefixedLog("blocking").Warn("blocking couldn't be disabled: ", disableErr) } } - - case <-ctx.Done(): - return - } + }() } + + err = evt.Bus().Subscribe(evt.BlockingStateChangedRemote, remoteHandler) + if err != nil { + return fmt.Errorf("failed to subscribe to %s: %w", evt.BlockingStateChangedRemote, err) + } + + go func() { + <-ctx.Done() + _ = evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, remoteHandler) + }() + + return nil } // RefreshLists triggers the refresh of all allow/denylists in the cache @@ -226,10 +223,7 @@ func (r *BlockingResolver) retrieveAllBlockingGroups() []string { // EnableBlocking enables the blocking against the denylists func (r *BlockingResolver) EnableBlocking(ctx context.Context) { r.internalEnableBlocking() - - if r.redisClient != nil { - r.redisClient.PublishEnabled(ctx, &redis.EnabledMessage{State: true}) - } + evt.Bus().Publish(evt.BlockingStateChanged, evt.BlockingState{Enabled: true}) } func (r *BlockingResolver) internalEnableBlocking() { @@ -246,9 +240,9 @@ func (r *BlockingResolver) internalEnableBlocking() { // DisableBlocking deactivates the blocking for a particular duration (or forever if 0). func (r *BlockingResolver) DisableBlocking(ctx context.Context, duration time.Duration, disableGroups []string) error { err := r.internalDisableBlocking(ctx, duration, disableGroups) - if err == nil && r.redisClient != nil { - r.redisClient.PublishEnabled(ctx, &redis.EnabledMessage{ - State: false, + if err == nil { + evt.Bus().Publish(evt.BlockingStateChanged, evt.BlockingState{ + Enabled: false, Duration: duration, Groups: disableGroups, }) diff --git a/resolver/blocking_resolver_test.go b/resolver/blocking_resolver_test.go index 64b61ff5..c327a111 100644 --- a/resolver/blocking_resolver_test.go +++ b/resolver/blocking_resolver_test.go @@ -10,10 +10,7 @@ import ( "github.com/0xERR0R/blocky/lists" "github.com/0xERR0R/blocky/log" . "github.com/0xERR0R/blocky/model" - "github.com/0xERR0R/blocky/redis" "github.com/0xERR0R/blocky/util" - "github.com/alicebob/miniredis/v2" - "github.com/creasty/defaults" "github.com/miekg/dns" . "github.com/onsi/ginkgo/v2" @@ -71,7 +68,7 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() { m = &mockResolver{} m.On("Resolve", mock.Anything).Return(&Response{Res: mockAnswer}, nil) - sut, err = NewBlockingResolver(ctx, sutConfig, nil, systemResolverBootstrap) + sut, err = NewBlockingResolver(ctx, sutConfig, systemResolverBootstrap) Expect(err).Should(Succeed()) sut.Next(m) }) @@ -112,7 +109,7 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() { Expect(err).Should(Succeed()) // recreate to trigger a reload - sut, err = NewBlockingResolver(ctx, sutConfig, nil, systemResolverBootstrap) + sut, err = NewBlockingResolver(ctx, sutConfig, systemResolverBootstrap) Expect(err).Should(Succeed()) Eventually(groupCnt, "1s").Should(HaveLen(2)) @@ -1145,7 +1142,7 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() { It("should return error", func() { _, err := NewBlockingResolver(ctx, config.Blocking{ BlockType: "wrong", - }, nil, systemResolverBootstrap) + }, systemResolverBootstrap) Expect(err).Should(HaveOccurred()) Expect(err.Error()).Should(ContainSubstring( @@ -1161,82 +1158,35 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() { Init: config.Init{Strategy: config.InitStrategyFailOnError}, }, BlockType: "zeroIp", - }, nil, systemResolverBootstrap) + }, systemResolverBootstrap) Expect(err).Should(HaveOccurred()) }) }) }) - Describe("Redis is configured", func() { - var redisServer *miniredis.Miniredis - var redisClient *redis.Client - var err error - JustBeforeEach(func() { - redisServer, err = miniredis.Run() + Describe("Remote blocking state via event bus", func() { + When("BlockingStateChangedRemote with enabled=true is received", func() { + It("should enable blocking", func() { + Expect(sut.DisableBlocking(ctx, 0, []string{})).Should(Succeed()) + Expect(sut.BlockingStatus().Enabled).Should(BeFalse()) - Expect(err).Should(Succeed()) - - var rcfg config.Redis - err = defaults.Set(&rcfg) - - Expect(err).Should(Succeed()) - rcfg.Address = redisServer.Addr() - redisClient, err = redis.New(context.TODO(), &rcfg) - - Expect(err).Should(Succeed()) - Expect(redisClient).ShouldNot(BeNil()) - sutConfig = config.Blocking{ - BlockType: "ZEROIP", - BlockTTL: config.Duration(time.Minute), - } - - sut, err = NewBlockingResolver(ctx, sutConfig, redisClient, systemResolverBootstrap) - Expect(err).Should(Succeed()) - }) - JustAfterEach(func() { - redisServer.Close() - }) - When("disable", func() { - It("should return disable", func() { - sut.EnableBlocking(context.TODO()) - - redisMockMsg := &redis.EnabledMessage{ - State: false, - } - redisClient.EnabledChannel <- redisMockMsg + Bus().Publish(BlockingStateChangedRemote, BlockingState{Enabled: true}) Eventually(func() bool { return sut.BlockingStatus().Enabled - }, "5s").Should(BeFalse()) + }).Should(BeTrue()) }) }) - When("disable", func() { - It("should return disable", func() { - sut.EnableBlocking(context.TODO()) - redisMockMsg := &redis.EnabledMessage{ - State: false, - Groups: []string{"unknown"}, - } - redisClient.EnabledChannel <- redisMockMsg + When("BlockingStateChangedRemote with enabled=false is received", func() { + It("should disable blocking", func() { + Bus().Publish(BlockingStateChangedRemote, BlockingState{ + Enabled: false, + Groups: []string{}, + }) Eventually(func() bool { return sut.BlockingStatus().Enabled - }, "5s").Should(BeTrue()) - }) - }) - When("enable", func() { - It("should return enable", func() { - err = sut.DisableBlocking(context.TODO(), time.Hour, []string{}) - Expect(err).Should(Succeed()) - - redisMockMsg := &redis.EnabledMessage{ - State: true, - } - redisClient.EnabledChannel <- redisMockMsg - - Eventually(func() bool { - return sut.BlockingStatus().Enabled - }, "5s").Should(BeTrue()) + }).Should(BeFalse()) }) }) }) diff --git a/resolver/caching_resolver.go b/resolver/caching_resolver.go index da530cbb..b3928819 100644 --- a/resolver/caching_resolver.go +++ b/resolver/caching_resolver.go @@ -9,15 +9,14 @@ import ( "sync/atomic" "time" + "github.com/0xERR0R/blocky/cache" "github.com/0xERR0R/blocky/config" "github.com/0xERR0R/blocky/evt" "github.com/0xERR0R/blocky/metrics" "github.com/0xERR0R/blocky/model" - "github.com/0xERR0R/blocky/redis" "github.com/0xERR0R/blocky/util" expirationcache "github.com/0xERR0R/expiration-cache" - "github.com/0xERR0R/blocky/cache" "github.com/0xERR0R/blocky/cache/prefetching" "github.com/miekg/dns" "github.com/prometheus/client_golang/prometheus" @@ -27,6 +26,9 @@ import ( const defaultCachingCleanUpInterval = 5 * time.Second +// CacheDecorator optionally wraps the result cache (e.g., with Redis sync). +type CacheDecorator func(cache.ExpiringCache[[]byte]) (cache.ExpiringCache[[]byte], error) + //nolint:gochecknoglobals var ( cacheHits = promauto.With(metrics.Reg).NewCounter( @@ -54,38 +56,38 @@ type CachingResolver struct { resultCache cache.ExpiringCache[[]byte] - redisClient *redis.Client - compiledExclusions []*regexp.Regexp } // NewCachingResolver creates a new resolver instance func NewCachingResolver(ctx context.Context, cfg config.Caching, - redis *redis.Client, + decorator CacheDecorator, ) (*CachingResolver, error) { - return newCachingResolver(ctx, cfg, redis, true) + return newCachingResolver(ctx, cfg, decorator, true) } func newCachingResolver(ctx context.Context, cfg config.Caching, - redis *redis.Client, + decorator CacheDecorator, emitMetricEvents bool, ) (*CachingResolver, error) { c := &CachingResolver{ - configurable: withConfig(&cfg), - typed: withType("caching"), - - redisClient: redis, + configurable: withConfig(&cfg), + typed: withType("caching"), emitMetricEvents: emitMetricEvents, } configureCaches(ctx, c, &cfg) err := configureExclusions(c, &cfg) - if c.redisClient != nil { - go c.redisSubscriber(ctx) - c.redisClient.GetRedisCache(ctx) + if decorator != nil { + decorated, err := decorator(c.resultCache) + if err != nil { + return nil, fmt.Errorf("failed to apply cache decorator: %w", err) + } + + c.resultCache = decorated } return c, err @@ -175,24 +177,6 @@ func (r *CachingResolver) reloadCacheEntry(ctx context.Context, cacheKey string) return &packed, r.adjustTTLs(response.Res.Answer) } -func (r *CachingResolver) redisSubscriber(ctx context.Context) { - ctx, logger := r.log(ctx) - - for { - select { - case rc := <-r.redisClient.CacheChannel: - if rc != nil { - logger.Debug("Received key from redis: ", rc.Key) - ttl := r.adjustTTLs(rc.Response.Res.Answer) - r.putInCache(ctx, rc.Key, rc.Response, ttl, false) - } - - case <-ctx.Done(): - return - } - } -} - // LogConfig implements `config.Configurable`. func (r *CachingResolver) LogConfig(logger *logrus.Entry) { r.cfg.LogConfig(logger) @@ -238,7 +222,7 @@ func (r *CachingResolver) Resolve(ctx context.Context, request *model.Request) ( if err == nil { cacheTTL := r.adjustTTLs(response.Res.Answer) - r.putInCache(ctx, cacheKey, response, cacheTTL, true) + r.putInCache(ctx, cacheKey, response, cacheTTL) } else { return nil, err } @@ -322,7 +306,7 @@ func isResponseCacheable(msg *dns.Msg) bool { } func (r *CachingResolver) putInCache( - ctx context.Context, cacheKey string, response *model.Response, ttl time.Duration, publish bool, + ctx context.Context, cacheKey string, response *model.Response, ttl time.Duration, ) { respCopy := response.Res.Copy() @@ -343,11 +327,6 @@ func (r *CachingResolver) putInCache( } } } - - if publish && r.redisClient != nil { - res := *respCopy - r.redisClient.PublishCache(cacheKey, &res) - } } // adjustTTLs calculates and returns the min TTL (considers also the min and max cache time) diff --git a/resolver/caching_resolver_test.go b/resolver/caching_resolver_test.go index 89c2f8dd..fd8bdb57 100644 --- a/resolver/caching_resolver_test.go +++ b/resolver/caching_resolver_test.go @@ -12,9 +12,7 @@ import ( . "github.com/0xERR0R/blocky/helpertest" "github.com/0xERR0R/blocky/log" . "github.com/0xERR0R/blocky/model" - "github.com/0xERR0R/blocky/redis" "github.com/0xERR0R/blocky/util" - "github.com/alicebob/miniredis/v2" "github.com/creasty/defaults" "github.com/miekg/dns" @@ -719,80 +717,6 @@ var _ = Describe("CachingResolver", func() { }) }) - Describe("Redis is configured", func() { - var ( - redisServer *miniredis.Miniredis - redisClient *redis.Client - redisConfig *config.Redis - err error - ) - BeforeEach(func() { - redisServer, err = miniredis.Run() - - Expect(err).Should(Succeed()) - - var rcfg config.Redis - err = defaults.Set(&rcfg) - - Expect(err).Should(Succeed()) - - rcfg.Address = redisServer.Addr() - redisConfig = &rcfg - redisClient, err = redis.New(context.TODO(), redisConfig) - - Expect(err).Should(Succeed()) - Expect(redisClient).ShouldNot(BeNil()) - }) - AfterEach(func() { - redisServer.Close() - }) - When("cache", func() { - JustBeforeEach(func() { - sutConfig = config.Caching{ - MaxCachingTime: config.Duration(time.Second * 10), - } - mockAnswer, _ = util.NewMsgWithAnswer("example.com.", 1000, A, "1.1.1.1") - - sut, _ = NewCachingResolver(ctx, sutConfig, redisClient) - m = &mockResolver{} - m.On("Resolve", mock.Anything).Return(&Response{Res: mockAnswer}, nil) - sut.Next(m) - }) - - It("put in redis", func() { - Expect(sut.Resolve(ctx, newRequest("example.com.", A))). - Should(HaveResponseType(ResponseTypeRESOLVED)) - - Eventually(func() []string { - return redisServer.DB(redisConfig.Database).Keys() - }).Should(HaveLen(1)) - }) - - It("load", func() { - request := newRequest("example2.com.", A) - domain := util.ExtractDomain(request.Req.Question[0]) - cacheKey := util.GenerateCacheKey(A, domain) - redisMockMsg := &redis.CacheMessage{ - Key: cacheKey, - Response: &Response{ - RType: ResponseTypeCACHED, - Reason: "MOCK_REDIS", - Res: mockAnswer, - }, - } - redisClient.CacheChannel <- redisMockMsg - - Eventually(sut.Resolve). - WithContext(ctx). - WithArguments(request). - Should( - SatisfyAll( - HaveResponseType(ResponseTypeCACHED), - HaveTTL(BeNumerically("<=", 10)), - )) - }) - }) - }) Context("isRequestCacheable", func() { var request *Request When("request is not cacheable", func() { diff --git a/resolver/parallel_best_resolver_test.go b/resolver/parallel_best_resolver_test.go index b4d9be54..225a0d09 100644 --- a/resolver/parallel_best_resolver_test.go +++ b/resolver/parallel_best_resolver_test.go @@ -316,7 +316,7 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() { } for _, v := range resolverCount { // should be 500 ± 50 - Expect(v).Should(BeNumerically("~", 500, 75)) + Expect(v).Should(BeNumerically("~", 500, 100)) } }) By("perform 100 request, error upstream's weight will be reduced", func() { @@ -329,7 +329,7 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() { By("Resolvers without errors should be selected often", func() { resolverCount := make(map[*UpstreamResolver]int) - for range 100 { + for range 1000 { resolvers := pickRandom(ctx, *sut.resolvers.Load(), parallelBestResolverCount) res1 := resolvers[0].resolver.(*UpstreamResolver) res2 := resolvers[1].resolver.(*UpstreamResolver) @@ -340,11 +340,11 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() { } for k, v := range resolverCount { if strings.Contains(k.String(), "wrong") { - // error resolvers: should be 0 - 10 - Expect(v).Should(BeNumerically("~", 0, 10)) + // error resolvers: should be rare + Expect(v).Should(BeNumerically("~", 0, 50)) } else { - // should be 90 ± 10 - Expect(v).Should(BeNumerically("~", 90, 10)) + // good resolvers get the vast majority of picks + Expect(v).Should(BeNumerically("~", 950, 75)) } } }) @@ -516,7 +516,7 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() { By("Resolvers without errors should be selected often", func() { resolverCount := make(map[*UpstreamResolver]int) - for range 200 { + for range 1000 { r := weightedRandom(ctx, *sut.resolvers.Load(), nil) res := r.resolver.(*UpstreamResolver) @@ -524,11 +524,11 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() { } for k, v := range resolverCount { if strings.Contains(k.String(), "wrong") { - // error resolvers: should be 0 - 10 - Expect(v).Should(BeNumerically("~", 0, 10)) + // error resolvers: should be rare + Expect(v).Should(BeNumerically("~", 0, 25)) } else { - // should be 90 ± 10 - Expect(v).Should(BeNumerically("~", 95, 20)) + // good resolvers get the vast majority of picks + Expect(v).Should(BeNumerically("~", 490, 50)) } } }) diff --git a/resolver/resolver_test.go b/resolver/resolver_test.go index 38ca54fb..d61978f3 100644 --- a/resolver/resolver_test.go +++ b/resolver/resolver_test.go @@ -123,7 +123,7 @@ var _ = Describe("Resolver", func() { }) When("'Name' is called", func() { It("should return resolver name", func() { - br, _ := NewBlockingResolver(ctx, config.Blocking{BlockType: "zeroIP"}, nil, systemResolverBootstrap) + br, _ := NewBlockingResolver(ctx, config.Blocking{BlockType: "zeroIP"}, systemResolverBootstrap) name := Name(br) Expect(name).Should(Equal("blocking")) }) diff --git a/server/server.go b/server/server.go index da79b655..512f42ea 100644 --- a/server/server.go +++ b/server/server.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "errors" "fmt" + "io" "net" "net/http" "runtime" @@ -12,6 +13,7 @@ import ( "strings" "time" + "github.com/0xERR0R/blocky/cache" "github.com/0xERR0R/blocky/config" "github.com/0xERR0R/blocky/log" "github.com/0xERR0R/blocky/metrics" @@ -20,6 +22,7 @@ import ( "github.com/0xERR0R/blocky/resolver" "github.com/0xERR0R/blocky/util" + goredis "github.com/go-redis/redis/v8" "github.com/google/uuid" "github.com/hashicorp/go-multierror" @@ -41,6 +44,7 @@ type Server struct { cfg *config.Config servers map[net.Listener]*httpServer + closers []io.Closer } func logger() *logrus.Entry { @@ -128,15 +132,24 @@ func NewServer(ctx context.Context, cfg *config.Config) (server *Server, err err return nil, fmt.Errorf("failed to create bootstrap resolver: %w", err) } - var redisClient *redis.Client + var redisConn *goredis.Client if cfg.Redis.IsEnabled() { - redisClient, err = redis.New(ctx, &cfg.Redis) - if err != nil && cfg.Redis.Required { - return nil, fmt.Errorf("failed to create required Redis client: %w", err) + redisConn, err = redis.New(ctx, &cfg.Redis) + if err != nil { + if cfg.Redis.Required { + return nil, fmt.Errorf("failed to create required Redis client: %w", err) + } + + logger().WithError(err).Warn("Redis is enabled but optional and could not be initialized, continuing without Redis") } } - queryResolver, queryError := createQueryResolver(ctx, cfg, bootstrap, redisClient) + redisResult, err := createRedisCacheDecorator(ctx, redisConn, cfg.Redis.Required) + if err != nil { + return nil, err + } + + queryResolver, queryError := createQueryResolver(ctx, cfg, bootstrap, redisResult.decorator) if queryError != nil { return nil, queryError } @@ -149,6 +162,14 @@ func NewServer(ctx context.Context, cfg *config.Config) (server *Server, err err servers: make(map[net.Listener]*httpServer), } + if redisResult.bridge != nil { + server.closers = append(server.closers, redisResult.bridge) + } + + if redisConn != nil { + server.closers = append(server.closers, redisConn) + } + server.printConfiguration() server.registerDNSHandlers(ctx) @@ -294,19 +315,55 @@ func createUDPServer(address string) (*dns.Server, error) { return createDNSServer("udp", address, nil) } +type redisBridgeResult struct { + decorator resolver.CacheDecorator + bridge *redis.EventBusBridge +} + +func createRedisCacheDecorator( + ctx context.Context, redisConn *goredis.Client, required bool, +) (*redisBridgeResult, error) { + if redisConn == nil { + return &redisBridgeResult{}, nil + } + + bridge, err := redis.NewEventBusBridge(ctx, redisConn) + if err != nil { + if required { + return nil, fmt.Errorf("failed to create required Redis event bridge: %w", err) + } + + logger().Warn("failed to create Redis event bridge: ", err) + } + + decorator := func(inner cache.ExpiringCache[[]byte]) (cache.ExpiringCache[[]byte], error) { + return cache.NewRedisExpiringByteCache(ctx, inner, redisConn, cache.RedisOptions[[]byte]{ + Prefix: "blocky:cache:", + Channel: "blocky_cache_sync", + }) + } + + return &redisBridgeResult{decorator: decorator, bridge: bridge}, nil +} + func createQueryResolver( ctx context.Context, cfg *config.Config, bootstrap *resolver.Bootstrap, - redisClient *redis.Client, + cacheDecorator resolver.CacheDecorator, ) (resolver.ChainedResolver, error) { upstreamTree, utErr := resolver.NewUpstreamTreeResolver(ctx, cfg.Upstreams, bootstrap) - blocking, blErr := resolver.NewBlockingResolver(ctx, cfg.Blocking, redisClient, bootstrap) + blocking, blErr := resolver.NewBlockingResolver(ctx, cfg.Blocking, bootstrap) clientNames, cnErr := resolver.NewClientNamesResolver(ctx, cfg.ClientLookup, cfg.Upstreams, bootstrap) queryLogging, qlErr := resolver.NewQueryLoggingResolver(ctx, cfg.QueryLog) condUpstream, cuErr := resolver.NewConditionalUpstreamResolver(ctx, cfg.Conditional, cfg.Upstreams, bootstrap) hostsFile, hfErr := resolver.NewHostsFileResolver(ctx, cfg.HostsFile, bootstrap) - cachingResolver, crErr := resolver.NewCachingResolver(ctx, cfg.Caching, redisClient) + decorator := cacheDecorator + if !cfg.Caching.IsEnabled() { + decorator = nil + } + + cachingResolver, crErr := resolver.NewCachingResolver(ctx, cfg.Caching, decorator) // Pass upstreamTree to DNSSEC resolver so it can query for DNSKEY/DS records dnssecResolver, dsErr := resolver.NewDNSSECResolver(ctx, cfg.DNSSEC, upstreamTree) @@ -429,6 +486,12 @@ func (s *Server) Start(ctx context.Context, errCh chan<- error) { func (s *Server) Stop(ctx context.Context) error { logger().Info("Stopping server") + for _, c := range s.closers { + if err := c.Close(); err != nil { + logger().Warn("failed to close resource: ", err) + } + } + for _, server := range s.dnsServers { if err := server.ShutdownContext(ctx); err != nil { return fmt.Errorf("stop %s listener failed: %w", server.Net, err)