From a1c9a01aaa8332888f732f01409aa7473e764433 Mon Sep 17 00:00:00 2001 From: yashnevatia Date: Wed, 16 Sep 2026 10:08:40 +0100 Subject: [PATCH 1/2] Add cache size for unknown requests in consensus handler --- libs/chainconsensus/handler.go | 14 ++++++++--- libs/chainconsensus/handler_test.go | 39 ++++++++++++++++++++++++++--- 2 files changed, 46 insertions(+), 7 deletions(-) diff --git a/libs/chainconsensus/handler.go b/libs/chainconsensus/handler.go index 9a1913367..d11551341 100644 --- a/libs/chainconsensus/handler.go +++ b/libs/chainconsensus/handler.go @@ -50,18 +50,20 @@ type handler struct { unknownRequestsResultByID map[string]*unknownRequest unknownRequestsOrderedByTimeout *list.List[*unknownRequest] unknownRequestTTL time.Duration + maxUnknownRequestsCacheSize int } -func NewHandler(lggr logger.Logger, poller Poller, metrics metrics.ConsensusMetrics, unknownRequestTTL time.Duration) Handler { - return newHandler(lggr, poller, metrics, unknownRequestTTL) +func NewHandler(lggr logger.Logger, poller Poller, metrics metrics.ConsensusMetrics, unknownRequestTTL time.Duration, maxUnknownRequestsCacheSize int) Handler { + return newHandler(lggr, poller, metrics, unknownRequestTTL, maxUnknownRequestsCacheSize) } -func newHandler(lggr logger.Logger, poller Poller, metrics metrics.ConsensusMetrics, unknownRequestTTL time.Duration) *handler { +func newHandler(lggr logger.Logger, poller Poller, metrics metrics.ConsensusMetrics, unknownRequestTTL time.Duration, maxUnknownRequestsCacheSize int) *handler { r := &handler{ requests: requests.NewStoreWithStatsCollector[*requestCtx](metrics), unknownRequestsResultByID: make(map[string]*unknownRequest), unknownRequestsOrderedByTimeout: list.New[*unknownRequest](), unknownRequestTTL: unknownRequestTTL, + maxUnknownRequestsCacheSize: maxUnknownRequestsCacheSize, poller: poller, metrics: metrics, } @@ -210,6 +212,12 @@ func (s *handler) completeRequest(id string, reply types.Reply) error { defer s.lock.Unlock() request := s.requests.Get(id) if request == nil { + if s.maxUnknownRequestsCacheSize > 0 && len(s.unknownRequestsResultByID) >= s.maxUnknownRequestsCacheSize { + if oldest := s.unknownRequestsOrderedByTimeout.Front(); oldest != nil { + delete(s.unknownRequestsResultByID, oldest.Value.ID) + s.unknownRequestsOrderedByTimeout.Remove(oldest) + } + } uRequest := &unknownRequest{ ID: id, ExpiresAt: time.Now().Add(s.unknownRequestTTL), diff --git a/libs/chainconsensus/handler_test.go b/libs/chainconsensus/handler_test.go index 483536e8e..287125089 100644 --- a/libs/chainconsensus/handler_test.go +++ b/libs/chainconsensus/handler_test.go @@ -26,7 +26,7 @@ import ( func TestGetRequestIDs(t *testing.T) { poller := mocks.NewPoller(t) poller.EXPECT().Enqueue(mock.Anything, mock.Anything) - handler := NewHandler(logger.Test(t), poller, test.GetConsensusMetrics(t), time.Second) + handler := NewHandler(logger.Test(t), poller, test.GetConsensusMetrics(t), time.Second, 1000) addRequestToHandler := func(t *testing.T, ctx context.Context, id string) { request := types.NewEventuallyConsistentRequest(id, nil) _, err := handler.Handle(ctx, request) @@ -72,7 +72,7 @@ func TestGetRequestIDs(t *testing.T) { func TestGetRequest(t *testing.T) { poller := mocks.NewPoller(t) poller.EXPECT().Enqueue(mock.Anything, mock.Anything).Maybe() - handler := NewHandler(logger.Test(t), poller, test.GetConsensusMetrics(t), time.Second) + handler := NewHandler(logger.Test(t), poller, test.GetConsensusMetrics(t), time.Second, 1000) addRequestToHandler := func(t *testing.T, ctx context.Context, id string) { request := types.NewAggregatableRequest(id, nil) _, err := handler.Handle(ctx, request) @@ -102,7 +102,7 @@ func TestGetRequest(t *testing.T) { func TestCompleteRequest(t *testing.T) { newHandler := func(t *testing.T, lggr logger.Logger, poller Poller) *handler { - h := newHandler(lggr, poller, test.GetConsensusMetrics(t), time.Second) + h := newHandler(lggr, poller, test.GetConsensusMetrics(t), time.Second, 1000) require.NoError(t, h.Start(t.Context())) t.Cleanup(func() { require.NoError(t, h.Close()) @@ -276,7 +276,7 @@ func TestCompleteRequest(t *testing.T) { func TestHandle(t *testing.T) { poller := mocks.NewPoller(t) - handler := NewHandler(logger.Test(t), poller, test.GetConsensusMetrics(t), time.Second) + handler := NewHandler(logger.Test(t), poller, test.GetConsensusMetrics(t), time.Second, 1000) require.NoError(t, handler.Start(t.Context())) t.Cleanup(func() { require.NoError(t, handler.Close()) @@ -309,3 +309,34 @@ func mustMarshalProto(t *testing.T, msg proto.Message) []byte { require.NoError(t, err) return data } + +func TestCompleteRequest_UnknownRequestsCacheEviction(t *testing.T) { + const maxCacheSize = 3 + h := newHandler(logger.Test(t), nil, test.GetConsensusMetrics(t), time.Minute, maxCacheSize) + + completeUnknown := func(t *testing.T, id string) { + require.NoError(t, h.CompleteProtoRequest(id, &types.RequestReport{ + Report: &types.RequestReport_EventuallyConsistent{EventuallyConsistent: []byte(id)}, + })) + } + + completeUnknown(t, "req-1") + completeUnknown(t, "req-2") + completeUnknown(t, "req-3") + + h.lock.RLock() + require.Len(t, h.unknownRequestsResultByID, maxCacheSize) + require.Contains(t, h.unknownRequestsResultByID, "req-1") + h.lock.RUnlock() + + // cache is full; completing one more unknown request must evict the oldest ("req-1") + completeUnknown(t, "req-4") + + h.lock.RLock() + require.Len(t, h.unknownRequestsResultByID, maxCacheSize) + require.NotContains(t, h.unknownRequestsResultByID, "req-1") + require.Contains(t, h.unknownRequestsResultByID, "req-2") + require.Contains(t, h.unknownRequestsResultByID, "req-3") + require.Contains(t, h.unknownRequestsResultByID, "req-4") + h.lock.RUnlock() +} From b245848a76dd55787b92629e32bd5e77b421a27c Mon Sep 17 00:00:00 2001 From: yashnevatia Date: Mon, 21 Sep 2026 12:25:42 +0100 Subject: [PATCH 2/2] add log line --- libs/chainconsensus/handler.go | 1 + 1 file changed, 1 insertion(+) diff --git a/libs/chainconsensus/handler.go b/libs/chainconsensus/handler.go index d11551341..3006d0eed 100644 --- a/libs/chainconsensus/handler.go +++ b/libs/chainconsensus/handler.go @@ -213,6 +213,7 @@ func (s *handler) completeRequest(id string, reply types.Reply) error { request := s.requests.Get(id) if request == nil { if s.maxUnknownRequestsCacheSize > 0 && len(s.unknownRequestsResultByID) >= s.maxUnknownRequestsCacheSize { + s.lggr.Warnf("unknown requests cache is full, evicting oldest request") if oldest := s.unknownRequestsOrderedByTimeout.Front(); oldest != nil { delete(s.unknownRequestsResultByID, oldest.Value.ID) s.unknownRequestsOrderedByTimeout.Remove(oldest)