Cache item support multiple Payload Hashes

.changeset/better-bears-glow.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/blocksize-capital-adapter': patch
+---
+
+fix for gRPC edge case

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

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,

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
 }

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

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))
+	})
+}

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)
+					}
 				}
 			}
 		}

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")
 	}
 }

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,
 	})
 }
 
@@ -269,10 +276,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 +292,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 +402,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)
 }

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) {