From 5bffa7ff991c41be80f8fef92efc3f23a1438400 Mon Sep 17 00:00:00 2001 From: Denis Chernov Date: Fri, 18 Sep 2026 13:04:03 -0700 Subject: [PATCH 1/3] fix gRPC edge-case --- packages/streams-adapter/cache/cache.go | 38 +++++++++++++-- packages/streams-adapter/cache/cache_test.go | 24 ++++++++-- packages/streams-adapter/common/types.go | 2 +- packages/streams-adapter/redcon/redcon.go | 6 ++- .../streams-adapter/redcon/redcon_test.go | 48 ++++++++++++++++++- packages/streams-adapter/server/server.go | 13 ++++- 6 files changed, 117 insertions(+), 14 deletions(-) diff --git a/packages/streams-adapter/cache/cache.go b/packages/streams-adapter/cache/cache.go index 0ef48ee69a8..e793e427f9b 100644 --- a/packages/streams-adapter/cache/cache.go +++ b/packages/streams-adapter/cache/cache.go @@ -107,21 +107,49 @@ func (c *Cache) SetNew(rawKey string, originalRequestData map[string]interface{} Status: types.StatusNew, Timestamp: time.Now(), OriginalRequestData: originalRequestData, - PayloadHash: payloadHash, + PayloadHashes: map[[32]byte]struct{}{payloadHash: {}}, } cacheItemsTotal.Inc() return true } -// PayloadHashByRawKey returns a copy of the payload hash stored for rawKey. -func (c *Cache) PayloadHashByRawKey(rawKey string) ([32]byte, bool) { +// AddPayloadHash registers an additional payload hash for an existing raw key. +// This allows the same cached feed to fan out observations to subscribers that +// joined with different request payloads (e.g. different overrides). Returns true +// if the hash was newly added. +func (c *Cache) AddPayloadHash(rawKey string, payloadHash [32]byte) bool { + c.mu.Lock() + defer c.mu.Unlock() + + item, ok := c.items[rawKey] + if !ok { + return false + } + if item.PayloadHashes == nil { + item.PayloadHashes = map[[32]byte]struct{}{payloadHash: {}} + return true + } + if _, exists := item.PayloadHashes[payloadHash]; exists { + return false + } + item.PayloadHashes[payloadHash] = struct{}{} + return true +} + +// PayloadHashesByRawKey returns all payload hashes registered for rawKey. +func (c *Cache) PayloadHashesByRawKey(rawKey string) ([][32]byte, bool) { c.mu.RLock() defer c.mu.RUnlock() + item, ok := c.items[rawKey] if !ok { - return [32]byte{}, false + return nil, false + } + hashes := make([][32]byte, 0, len(item.PayloadHashes)) + for h := range item.PayloadHashes { + hashes = append(hashes, h) } - return item.PayloadHash, true + return hashes, true } // SetTransformedKey transitions a "new" item to "learned" by recording the diff --git a/packages/streams-adapter/cache/cache_test.go b/packages/streams-adapter/cache/cache_test.go index 1bcec4fd0be..dc372efd517 100644 --- a/packages/streams-adapter/cache/cache_test.go +++ b/packages/streams-adapter/cache/cache_test.go @@ -204,10 +204,10 @@ func TestCache_SetNew_PreservesPayloadHash(t *testing.T) { defer c.Stop() payloadHash := [32]byte{1, 2, 3} require.True(t, c.SetNew("raw-key-with-hash", map[string]interface{}{"base": "ETH"}, payloadHash)) - require.Equal(t, payloadHash, c.Get("raw-key-with-hash").PayloadHash) - got, ok := c.PayloadHashByRawKey("raw-key-with-hash") + require.Contains(t, c.Get("raw-key-with-hash").PayloadHashes, payloadHash) + hashes, ok := c.PayloadHashesByRawKey("raw-key-with-hash") require.True(t, ok) - require.Equal(t, payloadHash, got) + require.Equal(t, [][32]byte{payloadHash}, hashes) } func TestCache_SetTransformedKey_StatusLearned(t *testing.T) { @@ -502,6 +502,24 @@ func TestCache_CleanupExpired_OrphanedPendingObs(t *testing.T) { assert.False(t, exists, "stale orphaned pending observation should be removed by cleanup") } +func TestCache_AddPayloadHash(t *testing.T) { + c := New(Config{TTL: time.Minute, CleanupInterval: time.Hour}) + defer c.Stop() + + hash1 := [32]byte{1, 2, 3} + hash2 := [32]byte{4, 5, 6} + + require.True(t, c.SetNew("raw-key", nil, hash1)) + require.True(t, c.AddPayloadHash("raw-key", hash2)) + require.False(t, c.AddPayloadHash("raw-key", hash1), "duplicate hash should be ignored") + + hashes, ok := c.PayloadHashesByRawKey("raw-key") + require.True(t, ok) + require.Len(t, hashes, 2) + require.Contains(t, hashes, hash1) + require.Contains(t, hashes, hash2) +} + func TestCache_DeterministicKeyOrdering(t *testing.T) { c := New(Config{ TTL: time.Minute, diff --git a/packages/streams-adapter/common/types.go b/packages/streams-adapter/common/types.go index 7aaad6d71a9..da8ac044af4 100644 --- a/packages/streams-adapter/common/types.go +++ b/packages/streams-adapter/common/types.go @@ -55,5 +55,5 @@ type CacheItem struct { Timestamp time.Time // last write time (used for TTL) OriginalAdapterKey string // JS adapter Redis key; populated once StatusActive OriginalRequestData map[string]interface{} // raw request body data from the first subscription request - PayloadHash [32]byte // SHA-256(adapter name || JSON request data) + PayloadHashes map[[32]byte]struct{} // all SHA-256 hashes (adapter name || JSON request data) registered for this raw key } diff --git a/packages/streams-adapter/redcon/redcon.go b/packages/streams-adapter/redcon/redcon.go index b34382e1325..968f6272770 100644 --- a/packages/streams-adapter/redcon/redcon.go +++ b/packages/streams-adapter/redcon/redcon.go @@ -243,8 +243,10 @@ func (s *RedconServer) handleEval(conn redcon.Conn, cmd redcon.Command) { if s.publisher != nil { if rawKeys, ok := s.cache.RawKeysByTransformed(transformedKey); ok { for _, rawKey := range rawKeys { - if payloadHash, ok := s.cache.PayloadHashByRawKey(rawKey); ok { - s.publisher.Publish(payloadHash, obs, ts) + if payloadHashes, ok := s.cache.PayloadHashesByRawKey(rawKey); ok { + for _, payloadHash := range payloadHashes { + s.publisher.Publish(payloadHash, obs, ts) + } } } } diff --git a/packages/streams-adapter/redcon/redcon_test.go b/packages/streams-adapter/redcon/redcon_test.go index 483606932de..f2d06aa8a12 100644 --- a/packages/streams-adapter/redcon/redcon_test.go +++ b/packages/streams-adapter/redcon/redcon_test.go @@ -12,6 +12,7 @@ import ( "github.com/stretchr/testify/require" "github.com/tidwall/redcon" + "streams-adapter/transmitter" ) // --------------------------------------------------------------------------- @@ -506,6 +507,51 @@ func TestHandleCommand_EvalTooFewArgs(t *testing.T) { require.Equal(t, "error", conn.writes[0].kind) errMsg, _ := conn.writes[0].value.(string) if !strings.Contains(errMsg, "wrong number of arguments") { - t.Errorf("error should mention wrong number of arguments, got: %s", errMsg) + t.Errorf("error should mention 'wrong number of arguments', got: %s", errMsg) + } +} + +func TestHandleCommand_Eval_FansOutToMultiplePayloadHashes(t *testing.T) { + c := cache.New(cache.Config{TTL: time.Minute, CleanupInterval: time.Hour}) + defer c.Stop() + + pub := transmitter.NewPublisher() + srv := New(Config{ + Addr: ":0", + Cache: c, + Publisher: pub, + Logger: slog.Default(), + }) + + rawKey := "endpoint=cryptolwba:from=btc:to=usd" + transformedKey := "base=btc:endpoint=cryptolwba:quote=usd" + hash1 := [32]byte{1, 2, 3} + hash2 := [32]byte{4, 5, 6} + + c.SetNew(rawKey, nil, hash1) + c.AddPayloadHash(rawKey, hash2) + c.SetTransformedKey(rawKey, transformedKey) + + ch1 := make(chan transmitter.Event, 1) + ch2 := make(chan transmitter.Event, 1) + pub.Subscribe(hash1, ch1) + pub.Subscribe(hash2, ch2) + + adapterKey := "prefix-adapter-endpoint-transport-" + transformedKey + value := `{"data":{"ask":"1"},"timestamps":{},"meta":{},"result":"1"}` + conn := newMockConn() + srv.handleCommand(conn, makeCmd("EVAL", "script", "1", adapterKey, value)) + + select { + case e := <-ch1: + require.Equal(t, hash1, e.PayloadHash) + case <-time.After(time.Second): + t.Fatal("subscriber on hash1 did not receive observation") + } + select { + case e := <-ch2: + require.Equal(t, hash2, e.PayloadHash) + case <-time.After(time.Second): + t.Fatal("subscriber on hash2 did not receive observation") } } diff --git a/packages/streams-adapter/server/server.go b/packages/streams-adapter/server/server.go index add12194568..aaf23d5f2a8 100644 --- a/packages/streams-adapter/server/server.go +++ b/packages/streams-adapter/server/server.go @@ -269,10 +269,14 @@ func (s *Server) cacheHandler(c *gin.Context) { Timestamp time.Time `json:"timestamp"` Observation *types.Observation `json:"observation,omitempty"` OriginalRequestData map[string]interface{} `json:"originalRequestData,omitempty"` - PayloadHash string `json:"payloadHash"` + PayloadHashes []string `json:"payloadHashes"` } entries := make([]entry, 0, len(items)) for key, item := range items { + hashes := make([]string, 0, len(item.PayloadHashes)) + for h := range item.PayloadHashes { + hashes = append(hashes, hex.EncodeToString(h[:])) + } entries = append(entries, entry{ Key: key, Status: item.Status, @@ -281,7 +285,7 @@ func (s *Server) cacheHandler(c *gin.Context) { Timestamp: item.Timestamp, Observation: item.Observation, OriginalRequestData: item.OriginalRequestData, - PayloadHash: hex.EncodeToString(item.PayloadHash[:]), + PayloadHashes: hashes, }) } @@ -391,9 +395,14 @@ func (s *Server) ResolveSubscription(data map[string]interface{}) (*types.Resolv // EnsureSubscription atomically creates a cache entry and starts provider // bootstrap for the first caller. Later HTTP or gRPC callers reuse that work. +// If the cache entry already exists but the caller's payload hash differs +// (different overrides, transport, etc.), the new hash is registered so the +// publisher can fan out observations to all matching subscribers. func (s *Server) EnsureSubscription(resolved *types.ResolvedSubscription) *types.CacheItem { if s.cache.SetNew(resolved.CacheKey, resolved.Data, resolved.PayloadHash) { go s.bootstrapSubscription(resolved.CacheKey, resolved.Params, resolved.Data) + } else { + s.cache.AddPayloadHash(resolved.CacheKey, resolved.PayloadHash) } return s.cache.Get(resolved.CacheKey) } From 2e94b47a814b23ddf684b5e78e274e1206ba2b39 Mon Sep 17 00:00:00 2001 From: Denis Chernov Date: Fri, 18 Sep 2026 15:07:59 -0700 Subject: [PATCH 2/3] add adapter version to /health endpoint --- packages/streams-adapter/main.go | 25 +++++++++++--- packages/streams-adapter/main_test.go | 33 +++++++++++++++++++ packages/streams-adapter/server/server.go | 11 +++++-- .../streams-adapter/server/server_test.go | 19 +++++++++++ 4 files changed, 81 insertions(+), 7 deletions(-) create mode 100644 packages/streams-adapter/main_test.go diff --git a/packages/streams-adapter/main.go b/packages/streams-adapter/main.go index ed71b28d652..c69a7c4f191 100644 --- a/packages/streams-adapter/main.go +++ b/packages/streams-adapter/main.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "log" "log/slog" @@ -21,8 +22,9 @@ import ( "streams-adapter/transmitter" ) -// waitForEAServer waits for the EA server to be ready before proceeding -func waitForEAServer(cfg *config.Config, logger *slog.Logger) { +// waitForEAServer waits for the EA server to be ready before proceeding. +// It returns the adapter version reported by the JS adapter's health endpoint. +func waitForEAServer(cfg *config.Config, logger *slog.Logger) string { eaURL := fmt.Sprintf("http://%s:%s%s/health", cfg.EAHost, cfg.EAPort, cfg.EABaseUrl) maxWaitTime := 60 * time.Second checkInterval := 500 * time.Millisecond @@ -46,9 +48,10 @@ func waitForEAServer(cfg *config.Config, logger *slog.Logger) { // Try to connect to the EA server health endpoint resp, err := client.Get(eaURL) if err == nil && resp.StatusCode == http.StatusOK { + version := readAdapterVersion(resp) resp.Body.Close() - logger.Info("EA server is ready", "elapsed", time.Since(startTime)) - return + logger.Info("EA server is ready", "elapsed", time.Since(startTime), "adapterVersion", version) + return version } if resp != nil { resp.Body.Close() @@ -57,6 +60,17 @@ func waitForEAServer(cfg *config.Config, logger *slog.Logger) { } } +// readAdapterVersion extracts the "version" field from the JS adapter health response. +func readAdapterVersion(resp *http.Response) string { + var health struct { + Version string `json:"version"` + } + if err := json.NewDecoder(resp.Body).Decode(&health); err != nil { + return "" + } + return health.Version +} + func main() { cfg := config.Load() logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) @@ -76,10 +90,11 @@ func main() { pub := transmitter.NewPublisher() // Wait for EA server to be ready before starting - waitForEAServer(cfg, logger) + adapterVersion := waitForEAServer(cfg, logger) // Initialize HTTP server httpServer := server.New(cfg, appCache, logger) + httpServer.SetAdapterVersion(adapterVersion) defer httpServer.Stop() // Initialize Redcon server diff --git a/packages/streams-adapter/main_test.go b/packages/streams-adapter/main_test.go new file mode 100644 index 00000000000..e35c34d2363 --- /dev/null +++ b/packages/streams-adapter/main_test.go @@ -0,0 +1,33 @@ +package main + +import ( + "io" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestReadAdapterVersion(t *testing.T) { + t.Run("returns version from response", func(t *testing.T) { + resp := &http.Response{ + Body: io.NopCloser(strings.NewReader(`{"message":"OK","version":"2.14.1"}`)), + } + require.Equal(t, "2.14.1", readAdapterVersion(resp)) + }) + + t.Run("returns empty for missing version", func(t *testing.T) { + resp := &http.Response{ + Body: io.NopCloser(strings.NewReader(`{"message":"OK"}`)), + } + require.Empty(t, readAdapterVersion(resp)) + }) + + t.Run("returns empty for invalid JSON", func(t *testing.T) { + resp := &http.Response{ + Body: io.NopCloser(strings.NewReader(`not json`)), + } + require.Empty(t, readAdapterVersion(resp)) + }) +} diff --git a/packages/streams-adapter/server/server.go b/packages/streams-adapter/server/server.go index aaf23d5f2a8..be9974aaa17 100644 --- a/packages/streams-adapter/server/server.go +++ b/packages/streams-adapter/server/server.go @@ -83,6 +83,7 @@ type Server struct { metricsForwarder *appMetrics.Forwarder ctx context.Context cancel context.CancelFunc + adapterVersion string } // New creates a new HTTP server @@ -150,6 +151,11 @@ func New(cfg *config.Config, cache *cache.Cache, logger *slog.Logger) *Server { return server } +// SetAdapterVersion records the JS adapter version reported by its health endpoint. +func (s *Server) SetAdapterVersion(version string) { + s.adapterVersion = version +} + // setupRoutes configures the HTTP routes func (s *Server) setupRoutes() { group := s.router.Group(s.config.EABaseUrl) @@ -252,8 +258,9 @@ func (s *Server) Stop() error { // healthHandler handles health check requests func (s *Server) healthHandler(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ - "status": "healthy", - "time": time.Now().UTC(), + "status": "healthy", + "time": time.Now().UTC(), + "adapterVersion": s.adapterVersion, }) } diff --git a/packages/streams-adapter/server/server_test.go b/packages/streams-adapter/server/server_test.go index 027b91b1079..6e0482d9431 100644 --- a/packages/streams-adapter/server/server_test.go +++ b/packages/streams-adapter/server/server_test.go @@ -96,6 +96,9 @@ func setCache(t *testing.T, params types.RequestParams, obs *types.Observation, } func TestHealthHandler(t *testing.T) { + // Reset version that may have been set by other tests. + testSrv.SetAdapterVersion("") + req := httptest.NewRequest(http.MethodGet, "/health", nil) w := httptest.NewRecorder() testSrv.router.ServeHTTP(w, req) @@ -105,6 +108,22 @@ func TestHealthHandler(t *testing.T) { var body map[string]interface{} require.NoError(t, json.Unmarshal(w.Body.Bytes(), &body)) require.Equal(t, "healthy", body["status"]) + require.Empty(t, body["adapterVersion"]) +} + +func TestHealthHandler_AdapterVersion(t *testing.T) { + testSrv.SetAdapterVersion("2.14.1") + t.Cleanup(func() { testSrv.SetAdapterVersion("") }) + + req := httptest.NewRequest(http.MethodGet, "/health", nil) + w := httptest.NewRecorder() + testSrv.router.ServeHTTP(w, req) + + require.Equal(t, http.StatusOK, w.Code) + + var body map[string]interface{} + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &body)) + require.Equal(t, "2.14.1", body["adapterVersion"]) } func TestAdapterHandler_BadRequest(t *testing.T) { From 71600de077ba192de92bcaa7202f81444b1125ad Mon Sep 17 00:00:00 2001 From: Denis Chernov Date: Mon, 21 Sep 2026 08:57:12 -0700 Subject: [PATCH 3/3] changeset --- .changeset/better-bears-glow.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/better-bears-glow.md diff --git a/.changeset/better-bears-glow.md b/.changeset/better-bears-glow.md new file mode 100644 index 00000000000..989e70a1e0b --- /dev/null +++ b/.changeset/better-bears-glow.md @@ -0,0 +1,5 @@ +--- +'@chainlink/blocksize-capital-adapter': patch +--- + +fix for gRPC edge case