From 58610af77c3e7dab05d5e3a56084c275f67f2b8a Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Thu, 5 Feb 2026 11:59:16 -0800
Subject: [PATCH 01/11] Replaced appendSlice and removed dead code

---
 libraries/corereader/scan_cache_test.go | 191 ------------------------
 libraries/corereader/slice_reader.go    |  14 +-
 2 files changed, 1 insertion(+), 204 deletions(-)

diff --git a/libraries/corereader/scan_cache_test.go b/libraries/corereader/scan_cache_test.go
index 7c05ad1..3afb9ac 100644
--- a/libraries/corereader/scan_cache_test.go
+++ b/libraries/corereader/scan_cache_test.go
@@ -2,7 +2,6 @@ package corereader
 
 import (
 	"os"
-	"sync"
 	"testing"
 	"time"
 )
@@ -226,193 +225,3 @@ func TestMultipleReadersShareCache(t *testing.T) {
 	})
 }
 
-func TestSharedMetadataUpdates(t *testing.T) {
-	scanCacheMu.Lock()
-	scanCache = make(map[string]*cachedScan)
-	scanCacheMu.Unlock()
-
-	testDataPath := getTestDataPath()
-	if _, err := os.Stat(testDataPath); err != nil {
-		t.Skipf("Test data not available: %v", err)
-	}
-
-	t.Run("MetadataWithinReaderIsShared", func(t *testing.T) {
-		// This test verifies that within a single reader, the shared metadata is consistent
-		reader1, err := NewSliceReaderWithOptions(testDataPath, QueryReaderOptions())
-		if err != nil {
-			t.Fatalf("Failed to create reader1: %v", err)
-		}
-		defer reader1.Close()
-
-		initialCount := reader1.sharedMetadata.getSliceCount()
-		t.Logf("Initial slice count: %d", initialCount)
-
-		// Simulate discovering a new slice (like coreindex would do)
-		newSlice := SliceInfo{
-			SliceNum:       uint32(initialCount),
-			StartBlock:     uint32(initialCount*10000 + 1),
-			EndBlock:       uint32(initialCount*10000 + 5000),
-			MaxBlock:       uint32((initialCount + 1) * 10000),
-			BlocksPerSlice: 10000,
-			Finalized:      false,
-			GlobMin:        999999,
-			GlobMax:        1000099,
-		}
-
-		// Append via shared metadata
-		reader1.sharedMetadata.appendSlice(newSlice)
-
-		// Verify the reader sees the update
-		newCount := reader1.sharedMetadata.getSliceCount()
-		if newCount != initialCount+1 {
-			t.Errorf("Reader slice count incorrect: got %d, expected %d", newCount, initialCount+1)
-		}
-
-		retrievedSlice := reader1.sharedMetadata.getSlice(initialCount)
-		if retrievedSlice.SliceNum != newSlice.SliceNum {
-			t.Errorf("Retrieved slice doesn't match: got %d, expected %d", retrievedSlice.SliceNum, newSlice.SliceNum)
-		}
-
-		t.Log("✓ SharedSliceMetadata within reader is consistent")
-	})
-}
-
-func TestConcurrentMetadataAccess(t *testing.T) {
-	scanCacheMu.Lock()
-	scanCache = make(map[string]*cachedScan)
-	scanCacheMu.Unlock()
-
-	testDataPath := getTestDataPath()
-	if _, err := os.Stat(testDataPath); err != nil {
-		t.Skipf("Test data not available: %v", err)
-	}
-
-	t.Run("ConcurrentReadsAndWrites", func(t *testing.T) {
-		reader1, err := NewSliceReaderWithOptions(testDataPath, QueryReaderOptions())
-		if err != nil {
-			t.Fatalf("Failed to create reader1: %v", err)
-		}
-		defer reader1.Close()
-
-		reader2, err := NewSliceReaderWithOptions(testDataPath, SyncReaderOptions())
-		if err != nil {
-			t.Fatalf("Failed to create reader2: %v", err)
-		}
-		defer reader2.Close()
-
-		var wg sync.WaitGroup
-		errors := make(chan error, 3)
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for i := 0; i < 100; i++ {
-				count := reader1.sharedMetadata.getSliceCount()
-				if count == 0 {
-					errors <- nil
-					return
-				}
-				_ = reader1.sharedMetadata.getSlice(0)
-			}
-		}()
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for i := 0; i < 100; i++ {
-				count := reader2.sharedMetadata.getSliceCount()
-				if count == 0 {
-					errors <- nil
-					return
-				}
-			}
-		}()
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			baseCount := reader1.sharedMetadata.getSliceCount()
-			for i := 0; i < 10; i++ {
-				newSlice := SliceInfo{
-					SliceNum:       uint32(baseCount + i),
-					StartBlock:     uint32((baseCount+i)*10000 + 1),
-					EndBlock:       uint32((baseCount+i)*10000 + 5000),
-					MaxBlock:       uint32((baseCount + i + 1) * 10000),
-					BlocksPerSlice: 10000,
-					Finalized:      false,
-					GlobMin:        uint64(1000000 + i*100),
-					GlobMax:        uint64(1000100 + i*100),
-				}
-				reader1.sharedMetadata.appendSlice(newSlice)
-				time.Sleep(time.Millisecond)
-			}
-		}()
-
-		wg.Wait()
-		close(errors)
-
-		for err := range errors {
-			if err != nil {
-				t.Errorf("Concurrent access error: %v", err)
-			}
-		}
-
-		t.Logf("Completed 200 reads + 10 writes with no races")
-	})
-}
-
-func TestSliceMapIndexLookup(t *testing.T) {
-	scanCacheMu.Lock()
-	scanCache = make(map[string]*cachedScan)
-	scanCacheMu.Unlock()
-
-	testDataPath := getTestDataPath()
-	if _, err := os.Stat(testDataPath); err != nil {
-		t.Skipf("Test data not available: %v", err)
-	}
-
-	t.Run("LookupRemainsValidAfterAppend", func(t *testing.T) {
-		reader, err := NewSliceReaderWithOptions(testDataPath, QueryReaderOptions())
-		if err != nil {
-			t.Fatalf("Failed to create reader: %v", err)
-		}
-		defer reader.Close()
-
-		initialCount := reader.sharedMetadata.getSliceCount()
-		if initialCount == 0 {
-			t.Skip("No slices available for testing")
-		}
-
-		firstSlice := reader.sharedMetadata.getSlice(0)
-		lookupKey := firstSlice.StartBlock / 10000
-
-		idx, exists := reader.sliceMapByBlock[lookupKey]
-		if !exists {
-			t.Fatalf("Slice map doesn't contain key %d", lookupKey)
-		}
-
-		lookupSlice := reader.sharedMetadata.getSlice(idx)
-		if lookupSlice.SliceNum != firstSlice.SliceNum {
-			t.Errorf("Initial lookup failed: got slice %d, expected %d", lookupSlice.SliceNum, firstSlice.SliceNum)
-		}
-
-		newSlice := SliceInfo{
-			SliceNum:       uint32(initialCount),
-			StartBlock:     uint32(initialCount*10000 + 1),
-			EndBlock:       uint32(initialCount*10000 + 5000),
-			MaxBlock:       uint32((initialCount + 1) * 10000),
-			BlocksPerSlice: 10000,
-			Finalized:      false,
-			GlobMin:        999999,
-			GlobMax:        1000099,
-		}
-		reader.sharedMetadata.appendSlice(newSlice)
-
-		lookupSliceAfter := reader.sharedMetadata.getSlice(idx)
-		if lookupSliceAfter.SliceNum != firstSlice.SliceNum {
-			t.Errorf("Lookup invalidated after append: got slice %d, expected %d", lookupSliceAfter.SliceNum, firstSlice.SliceNum)
-		}
-
-		t.Logf("Slice map lookup remained valid after append (capacity-based stability)")
-	})
-}
diff --git a/libraries/corereader/slice_reader.go b/libraries/corereader/slice_reader.go
index 66d002c..85ddde8 100644
--- a/libraries/corereader/slice_reader.go
+++ b/libraries/corereader/slice_reader.go
@@ -75,18 +75,6 @@ func (sm *SharedSliceMetadata) getSlices() []SliceInfo {
 	return result
 }
 
-func (sm *SharedSliceMetadata) appendSlice(slice SliceInfo) int {
-	sm.mu.Lock()
-	defer sm.mu.Unlock()
-	idx := len(sm.slices)
-	sm.slices = append(sm.slices, slice)
-	if sm.sliceIndex == nil {
-		sm.sliceIndex = make(map[uint32]int)
-	}
-	sm.sliceIndex[slice.SliceNum] = idx
-	return idx
-}
-
 func (sm *SharedSliceMetadata) updateSlice(idx int, slice SliceInfo) {
 	sm.mu.Lock()
 	defer sm.mu.Unlock()
@@ -1785,7 +1773,7 @@ func (sr *SliceReader) discoverSliceForGlob(glob uint64, knownSlices []SliceInfo
 				GlobMax:        globMax,
 			}
 
-			sr.sharedMetadata.appendSlice(newSlice)
+			sr.sharedMetadata.addSlice(newSlice)
 
 			return &newSlice, nil
 		}

From f3824e67e8ef85a624fbb6b2fc8d210907841b8f Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Tue, 10 Feb 2026 09:36:13 -0800
Subject: [PATCH 02/11] Fixed ABI decoding path

---
 libraries/abicache/reader.go | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/libraries/abicache/reader.go b/libraries/abicache/reader.go
index 06b4418..f580ba5 100644
--- a/libraries/abicache/reader.go
+++ b/libraries/abicache/reader.go
@@ -12,7 +12,6 @@ import (
 	"syscall"
 
 	goeosio "github.com/greymass/go-eosio/pkg/chain"
-	"github.com/greymass/roborovski/libraries/chain"
 )
 
 type Reader struct {
@@ -198,9 +197,8 @@ func (r *Reader) Decode(contract, action uint64, data []byte, atBlock uint32) (m
 		cachedABI = &abi
 	}
 
-	actionName := chain.NameToString(action)
 	reader := bytes.NewReader(data)
-	decoded, err := cachedABI.Decode(reader, actionName)
+	decoded, err := cachedABI.DecodeAction(reader, goeosio.Name(action))
 	if err != nil {
 		return nil, fmt.Errorf("failed to decode action: %w", err)
 	}

From 86da426a828c231704bc8e0178cdfc3e1b912ebc Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Wed, 11 Feb 2026 15:47:13 -0800
Subject: [PATCH 03/11] Surfacing resource usage across types and new /usage
 endpoint

---
 libraries/actionstream/client.go              |  40 ++--
 libraries/actionstream/client_test.go         |  20 +-
 libraries/chain/action.go                     |   2 +
 libraries/corereader/shared.go                |  20 +-
 libraries/corereader/types.go                 |  14 ++
 services/actionindex/cmd/actionindex/main.go  |   2 +
 services/actionindex/internal/broadcaster.go  |   6 +-
 services/actionindex/internal/openapi.go      |   1 +
 services/actionindex/internal/openapi.yaml    | 173 ++++++++++++++++++
 .../actionindex/internal/rpc_account_usage.go | 138 ++++++++++++++
 .../actionindex/internal/stream_catchup.go    |  14 +-
 services/actionindex/internal/stream_tcp.go   |   6 +-
 services/actionindex/internal/sync.go         |  33 ++--
 13 files changed, 421 insertions(+), 48 deletions(-)
 create mode 100644 services/actionindex/internal/rpc_account_usage.go

diff --git a/libraries/actionstream/client.go b/libraries/actionstream/client.go
index e72d9fd..9649401 100644
--- a/libraries/actionstream/client.go
+++ b/libraries/actionstream/client.go
@@ -51,13 +51,15 @@ func DefaultClientConfig() ClientConfig {
 }
 
 type Action struct {
-	GlobalSeq  uint64
-	BlockNum   uint32
-	BlockTime  uint32
-	Contract   string
-	Action     string
-	Receiver   string
-	ActionData []byte
+	GlobalSeq     uint64
+	BlockNum      uint32
+	BlockTime     uint32
+	Contract      string
+	Action        string
+	Receiver      string
+	ActionData    []byte
+	CpuUsageUs    uint32
+	NetUsageWords uint32
 }
 
 type Filter struct {
@@ -490,7 +492,7 @@ func (c *Client) writeMessage(w io.Writer, msgType uint8, payload []byte) error
 }
 
 func (c *Client) decodeAction(payload []byte) (Action, error) {
-	if len(payload) < 40 {
+	if len(payload) < 48 {
 		return Action{}, errors.New("action payload too short")
 	}
 
@@ -500,20 +502,24 @@ func (c *Client) decodeAction(payload []byte) (Action, error) {
 	contract := binary.LittleEndian.Uint64(payload[16:24])
 	actionName := binary.LittleEndian.Uint64(payload[24:32])
 	receiver := binary.LittleEndian.Uint64(payload[32:40])
+	cpuUsageUs := binary.LittleEndian.Uint32(payload[40:44])
+	netUsageWords := binary.LittleEndian.Uint32(payload[44:48])
 
 	var actionData []byte
-	if len(payload) > 40 {
-		actionData = payload[40:]
+	if len(payload) > 48 {
+		actionData = payload[48:]
 	}
 
 	return Action{
-		GlobalSeq:  globalSeq,
-		BlockNum:   blockNum,
-		BlockTime:  blockTime,
-		Contract:   chain.NameToString(contract),
-		Action:     chain.NameToString(actionName),
-		Receiver:   chain.NameToString(receiver),
-		ActionData: actionData,
+		GlobalSeq:     globalSeq,
+		BlockNum:      blockNum,
+		BlockTime:     blockTime,
+		Contract:      chain.NameToString(contract),
+		Action:        chain.NameToString(actionName),
+		Receiver:      chain.NameToString(receiver),
+		ActionData:    actionData,
+		CpuUsageUs:    cpuUsageUs,
+		NetUsageWords: netUsageWords,
 	}, nil
 }
 
diff --git a/libraries/actionstream/client_test.go b/libraries/actionstream/client_test.go
index 9f5e960..37e8dc6 100644
--- a/libraries/actionstream/client_test.go
+++ b/libraries/actionstream/client_test.go
@@ -28,13 +28,15 @@ func TestNewClient(t *testing.T) {
 func TestDecodeAction(t *testing.T) {
 	client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig())
 
-	payload := make([]byte, 40)
+	payload := make([]byte, 48)
 	binary.LittleEndian.PutUint64(payload[0:8], 12345)
 	binary.LittleEndian.PutUint32(payload[8:12], 1000)
 	binary.LittleEndian.PutUint32(payload[12:16], 1700000000)
 	binary.LittleEndian.PutUint64(payload[16:24], 0x5530EA033C80A555)
 	binary.LittleEndian.PutUint64(payload[24:32], 0x00000000A89C6360)
 	binary.LittleEndian.PutUint64(payload[32:40], 0x5530EA033C80A555)
+	binary.LittleEndian.PutUint32(payload[40:44], 500)
+	binary.LittleEndian.PutUint32(payload[44:48], 20)
 
 	action, err := client.decodeAction(payload)
 	if err != nil {
@@ -50,19 +52,27 @@ func TestDecodeAction(t *testing.T) {
 	if action.BlockTime != 1700000000 {
 		t.Errorf("expected BlockTime 1700000000, got %d", action.BlockTime)
 	}
+	if action.CpuUsageUs != 500 {
+		t.Errorf("expected CpuUsageUs 500, got %d", action.CpuUsageUs)
+	}
+	if action.NetUsageWords != 20 {
+		t.Errorf("expected NetUsageWords 20, got %d", action.NetUsageWords)
+	}
 }
 
 func TestDecodeActionWithData(t *testing.T) {
 	client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig())
 
-	payload := make([]byte, 48)
+	payload := make([]byte, 56)
 	binary.LittleEndian.PutUint64(payload[0:8], 100)
 	binary.LittleEndian.PutUint32(payload[8:12], 500)
 	binary.LittleEndian.PutUint32(payload[12:16], 1600000000)
 	binary.LittleEndian.PutUint64(payload[16:24], 0)
 	binary.LittleEndian.PutUint64(payload[24:32], 0)
 	binary.LittleEndian.PutUint64(payload[32:40], 0)
-	copy(payload[40:], []byte("testdata"))
+	binary.LittleEndian.PutUint32(payload[40:44], 0)
+	binary.LittleEndian.PutUint32(payload[44:48], 0)
+	copy(payload[48:], []byte("testdata"))
 
 	action, err := client.decodeAction(payload)
 	if err != nil {
@@ -634,7 +644,7 @@ func TestDialReceivesBatch(t *testing.T) {
 		payload := make([]byte, length-1)
 		conn.Read(payload)
 
-		actionPayload := make([]byte, 40)
+		actionPayload := make([]byte, 48)
 		binary.LittleEndian.PutUint64(actionPayload[0:8], 999)
 		binary.LittleEndian.PutUint32(actionPayload[8:12], 100)
 		binary.LittleEndian.PutUint32(actionPayload[12:16], 1700000000)
@@ -822,7 +832,7 @@ func TestRecvLoopReceivesActions(t *testing.T) {
 		close(serverReady)
 
 		for i := uint64(0); i < 3; i++ {
-			actionPayload := make([]byte, 40)
+			actionPayload := make([]byte, 48)
 			binary.LittleEndian.PutUint64(actionPayload[0:8], 1000+i)
 			binary.LittleEndian.PutUint32(actionPayload[8:12], uint32(100+i))
 			binary.LittleEndian.PutUint32(actionPayload[12:16], 1700000000)
diff --git a/libraries/chain/action.go b/libraries/chain/action.go
index b7bfe4b..77d2d49 100644
--- a/libraries/chain/action.go
+++ b/libraries/chain/action.go
@@ -66,6 +66,8 @@ type ActionTrace struct {
 	ProducerBlockID  string         `json:"producer_block_id"`
 	AccountRAMDeltas []AccountDelta `json:"account_ram_deltas"`
 
+	CpuUsageUs    uint32 `json:"-"`
+	NetUsageWords uint32 `json:"-"`
 	GlobalSeqUint64 uint64 `json:"-"`
 }
 
diff --git a/libraries/corereader/shared.go b/libraries/corereader/shared.go
index 7b8a709..5bea3db 100644
--- a/libraries/corereader/shared.go
+++ b/libraries/corereader/shared.go
@@ -936,7 +936,7 @@ func (cat *compressedActionTrace) At(blkData *blockData, blockNum uint32, glob u
 		AbiSequence:    cat.AbiSequence,
 	}
 
-	return &chain.ActionTrace{
+	at := &chain.ActionTrace{
 		ActionOrdinal:    cat.ActionOrdinal,
 		CreatorAO:        cat.CreatorAO,
 		ClosestUAAO:      cat.ClosestUAAO,
@@ -952,6 +952,14 @@ func (cat *compressedActionTrace) At(blkData *blockData, blockNum uint32, glob u
 		AccountRAMDeltas: ard,
 		GlobalSeqUint64:  glob,
 	}
+
+	if int(cat.TrxIDIndex) < len(blkData.TrxMetaInBlock) {
+		meta := &blkData.TrxMetaInBlock[cat.TrxIDIndex]
+		at.CpuUsageUs = meta.CpuUsageUs
+		at.NetUsageWords = meta.NetUsageWords
+	}
+
+	return at
 }
 
 func (cat *compressedActionTrace) CanonicalAt(blkData *blockData, glob uint64) CanonicalAction {
@@ -961,7 +969,7 @@ func (cat *compressedActionTrace) CanonicalAt(blkData *blockData, glob uint64) C
 		authAccountIndexes[i] = auth.AccountIndex
 	}
 
-	return CanonicalAction{
+	ca := CanonicalAction{
 		ActionOrdinal:      cat.ActionOrdinal,
 		CreatorAO:          cat.CreatorAO,
 		ReceiverUint64:     blkData.NamesInBlock[cat.ReceiverIndex],
@@ -972,6 +980,14 @@ func (cat *compressedActionTrace) CanonicalAt(blkData *blockData, glob uint64) C
 		ContractUint64:     blkData.NamesInBlock[cat.ContractNameIndex],
 		ActionUint64:       blkData.NamesInBlock[cat.ActionNameIndex],
 	}
+
+	if int(cat.TrxIDIndex) < len(blkData.TrxMetaInBlock) {
+		meta := &blkData.TrxMetaInBlock[cat.TrxIDIndex]
+		ca.CpuUsageUs = meta.CpuUsageUs
+		ca.NetUsageWords = meta.NetUsageWords
+	}
+
+	return ca
 }
 
 func catFromBytes(inBytes []byte) *compressedActionTrace {
diff --git a/libraries/corereader/types.go b/libraries/corereader/types.go
index 64d6ab6..d67d675 100644
--- a/libraries/corereader/types.go
+++ b/libraries/corereader/types.go
@@ -23,6 +23,8 @@ type CanonicalAction struct {
 	TrxIndex           uint32
 	ContractUint64     uint64
 	ActionUint64       uint64
+	CpuUsageUs         uint32
+	NetUsageWords      uint32
 }
 
 // RawBlock contains unfiltered block data as read from storage.
@@ -107,6 +109,18 @@ func (b *Block) SetRawBlock(raw *RawBlock) {
 	b.rawBlock = raw
 }
 
+func (b *Block) GetResourceUsage(globalSeq uint64) (cpuUs, netWords uint32) {
+	if b.rawBlock == nil {
+		return 0, 0
+	}
+	for i := range b.rawBlock.Actions {
+		if b.rawBlock.Actions[i].GlobalSeqUint64 == globalSeq {
+			return b.rawBlock.Actions[i].CpuUsageUs, b.rawBlock.Actions[i].NetUsageWords
+		}
+	}
+	return 0, 0
+}
+
 func (b *Block) GetTransactionID(trxIndex uint32) [32]byte {
 	if b.rawBlock == nil || b.rawBlock.rawData == nil {
 		return [32]byte{}
diff --git a/services/actionindex/cmd/actionindex/main.go b/services/actionindex/cmd/actionindex/main.go
index 282ef6a..ec66d7c 100644
--- a/services/actionindex/cmd/actionindex/main.go
+++ b/services/actionindex/cmd/actionindex/main.go
@@ -476,6 +476,8 @@ func handleRPC(
 		internal.HandleGetActions(cfg, store, indexes, reader, abiReader, w, r)
 	case strings.HasPrefix(path, "/account/") && strings.HasSuffix(path, "/activity"):
 		internal.HandleAccountActivity(cfg, store, indexes, reader, abiReader, w, r)
+	case strings.HasPrefix(path, "/account/") && strings.HasSuffix(path, "/usage"):
+		internal.HandleAccountUsage(cfg, store, indexes, reader, abiReader, w, r)
 	case strings.HasPrefix(path, "/account/") && strings.HasSuffix(path, "/log"):
 		internal.HandleAccountLog(cfg, store, indexes, reader, abiReader, w, r)
 	case strings.HasPrefix(path, "/account/") && strings.HasSuffix(path, "/stats"):
diff --git a/services/actionindex/internal/broadcaster.go b/services/actionindex/internal/broadcaster.go
index f97e1e3..9c4efe3 100644
--- a/services/actionindex/internal/broadcaster.go
+++ b/services/actionindex/internal/broadcaster.go
@@ -14,8 +14,10 @@ type StreamedAction struct {
 	BlockTime  uint32
 	Contract   uint64
 	Action     uint64
-	Receiver   uint64
-	ActionData []byte
+	Receiver      uint64
+	ActionData    []byte
+	CpuUsageUs    uint32
+	NetUsageWords uint32
 }
 
 type ActionBroadcaster struct {
diff --git a/services/actionindex/internal/openapi.go b/services/actionindex/internal/openapi.go
index c26c3a8..0145635 100644
--- a/services/actionindex/internal/openapi.go
+++ b/services/actionindex/internal/openapi.go
@@ -27,6 +27,7 @@ func InitOpenAPI(version string) error {
 var registeredRoutes = map[string][]string{
 	"/v1/history/get_actions":     {"GET", "POST"},
 	"/account/{account}/activity": {"GET"},
+	"/account/{account}/usage":    {"GET"},
 	"/account/{account}/log":      {"GET"},
 	"/account/{account}/stats":    {"GET"},
 }
diff --git a/services/actionindex/internal/openapi.yaml b/services/actionindex/internal/openapi.yaml
index dd77bc8..a866f0d 100644
--- a/services/actionindex/internal/openapi.yaml
+++ b/services/actionindex/internal/openapi.yaml
@@ -242,6 +242,102 @@ paths:
               schema:
                 $ref: "#/components/schemas/AccountActivityErrorResponse"
 
+  /account/{account}/usage:
+    get:
+      operationId: getAccountUsage
+      summary: Get account resource usage
+      description: Retrieve network resource usage for an account's actions with cursor-based pagination. Results are grouped by transaction, with CPU/NET usage at the transaction level and individual actions with their RAM deltas nested within. Supports the same filtering parameters as /activity.
+      tags:
+        - Account History
+      parameters:
+        - name: account
+          in: path
+          required: true
+          schema:
+            type: string
+          description: Account name
+        - name: limit
+          in: query
+          required: false
+          schema:
+            type: integer
+            default: 100
+            maximum: 1000
+          description: Max results (1-1000)
+        - name: order
+          in: query
+          required: false
+          schema:
+            type: string
+            enum: ["asc", "desc"]
+            default: "desc"
+          description: Sort order
+        - name: cursor
+          in: query
+          required: false
+          schema:
+            type: string
+          description: Pagination cursor
+        - name: contract
+          in: query
+          required: false
+          schema:
+            type: string
+          description: Filter by contract
+        - name: action
+          in: query
+          required: false
+          schema:
+            type: string
+          description: Filter by action (requires contract)
+        - name: date
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date
+          description: Single date filter (YYYY-MM-DD)
+        - name: start_date
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date
+          description: Date range start (YYYY-MM-DD)
+        - name: end_date
+          in: query
+          required: false
+          schema:
+            type: string
+            format: date
+          description: Date range end (YYYY-MM-DD)
+        - name: trace
+          in: query
+          required: false
+          schema:
+            type: boolean
+            default: false
+          description: Include query timing trace
+      responses:
+        "200":
+          description: Usage entries found
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AccountUsageResponse"
+        "400":
+          description: Bad request
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AccountActivityErrorResponse"
+        "503":
+          description: Service unavailable (syncing)
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AccountActivityErrorResponse"
+
   /account/{account}/log:
     get:
       operationId: getAccountLog
@@ -900,6 +996,83 @@ components:
             message:
               type: string
 
+    AccountUsageResponse:
+      type: object
+      required:
+        - results
+      properties:
+        results:
+          type: array
+          items:
+            $ref: "#/components/schemas/UsageTransaction"
+        next_cursor:
+          type: string
+          description: Cursor for next page
+        prev_cursor:
+          type: string
+          description: Cursor for previous page
+        trace:
+          type: object
+          description: Query trace information (if requested)
+
+    UsageTransaction:
+      type: object
+      description: Resource usage grouped by transaction. CPU and NET are per-transaction values; all actions within the same transaction share them.
+      required:
+        - trx_id
+        - block_num
+        - block_time
+        - cpu_usage_us
+        - net_usage_words
+        - actions
+      properties:
+        trx_id:
+          type: string
+        block_num:
+          type: integer
+          format: uint32
+        block_time:
+          type: string
+        cpu_usage_us:
+          type: integer
+          format: uint32
+          description: CPU usage in microseconds (per transaction)
+        net_usage_words:
+          type: integer
+          format: uint32
+          description: Network usage in 8-byte words (per transaction)
+        actions:
+          type: array
+          items:
+            $ref: "#/components/schemas/UsageAction"
+
+    UsageAction:
+      type: object
+      description: Individual action within a transaction
+      required:
+        - global_action_seq
+        - contract
+        - action
+        - account_ram_deltas
+      properties:
+        global_action_seq:
+          type: integer
+          format: uint64
+        contract:
+          type: string
+        action:
+          type: string
+        account_ram_deltas:
+          type: array
+          items:
+            type: object
+            properties:
+              account:
+                type: string
+              delta:
+                type: integer
+                format: int64
+
     AccountLogResponse:
       type: object
       required:
diff --git a/services/actionindex/internal/rpc_account_usage.go b/services/actionindex/internal/rpc_account_usage.go
new file mode 100644
index 0000000..0ef76c2
--- /dev/null
+++ b/services/actionindex/internal/rpc_account_usage.go
@@ -0,0 +1,138 @@
+package internal
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/greymass/roborovski/libraries/abicache"
+	"github.com/greymass/roborovski/libraries/chain"
+	"github.com/greymass/roborovski/libraries/corereader"
+	"github.com/greymass/roborovski/libraries/logger"
+	"github.com/greymass/roborovski/libraries/querytrace"
+)
+
+type UsageAction struct {
+	GlobalActionSeq  uint64               `json:"global_action_seq"`
+	Contract         string               `json:"contract"`
+	Action           string               `json:"action"`
+	AccountRAMDeltas []chain.AccountDelta `json:"account_ram_deltas"`
+}
+
+type UsageTransaction struct {
+	TrxID         string        `json:"trx_id"`
+	BlockNum      uint32        `json:"block_num"`
+	BlockTime     string        `json:"block_time"`
+	CpuUsageUs    uint32        `json:"cpu_usage_us"`
+	NetUsageWords uint32        `json:"net_usage_words"`
+	Actions       []UsageAction `json:"actions"`
+}
+
+type AccountUsageResponse struct {
+	Results    []UsageTransaction      `json:"results"`
+	NextCursor string                  `json:"next_cursor,omitempty"`
+	PrevCursor string                  `json:"prev_cursor,omitempty"`
+	Trace      *querytrace.TraceOutput `json:"trace,omitempty"`
+}
+
+func HandleAccountUsage(
+	cfg *Config,
+	store *Store,
+	indexes ActionIndexer,
+	reader corereader.Reader,
+	abiReader *abicache.Reader,
+	w http.ResponseWriter,
+	r *http.Request,
+) {
+	startTime := time.Now()
+
+	q, ok := newAccountQuery(cfg, indexes, reader, abiReader, w, r, "/usage", "account_usage")
+	if !ok {
+		return
+	}
+
+	if !q.execute() {
+		return
+	}
+
+	results, ok := q.fetchUsageEntries()
+	if !ok {
+		return
+	}
+
+	response := AccountUsageResponse{
+		Results:    results,
+		NextCursor: q.nextCursor,
+		PrevCursor: q.prevCursor,
+		Trace:      q.finalize(len(results)),
+	}
+
+	writeJSON(w, response)
+
+	logger.Printf("timing", "account_usage account=%s contract=%s action=%s date=%s results=%d duration=%v",
+		q.req.accountName, q.req.Contract, q.req.Action, q.req.Date, len(results), time.Since(startTime))
+}
+
+func (q *accountQuery) fetchUsageEntries() ([]UsageTransaction, bool) {
+	if len(q.selectedSeqs) == 0 {
+		return []UsageTransaction{}, true
+	}
+
+	var fetchStart time.Time
+	var cacheHitsBefore, cacheMissesBefore uint64
+	if q.trace.Enabled() {
+		fetchStart = time.Now()
+		cacheHitsBefore, cacheMissesBefore, _, _ = q.reader.GetBlockCacheStats()
+	}
+
+	results, timings, err := fetchUsageByGlobalSeqs(q.reader, q.selectedSeqs)
+	if err != nil {
+		logger.Printf("error", "account usage query failed for account=%s contract=%s action=%s: %v",
+			q.req.accountName, q.req.Contract, q.req.Action, err)
+		writeAccountActivityError(q.w, "internal_error", "failed to fetch actions", http.StatusInternalServerError)
+		return nil, false
+	}
+
+	q.timings = timings
+	q.addFetchTrace(fetchStart, cacheHitsBefore, cacheMissesBefore, len(results))
+
+	return results, true
+}
+
+func fetchUsageByGlobalSeqs(reader corereader.Reader, globalSeqs []uint64) ([]UsageTransaction, *corereader.FetchTimings, error) {
+	if len(globalSeqs) == 0 {
+		return []UsageTransaction{}, nil, nil
+	}
+
+	actions, timings, err := getActionsByGlobalSeqs(reader, globalSeqs)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var results []UsageTransaction
+	trxIndex := make(map[string]int)
+
+	for i, at := range actions {
+		action := UsageAction{
+			GlobalActionSeq:  globalSeqs[i],
+			Contract:         at.Act.Account,
+			Action:           at.Act.Name,
+			AccountRAMDeltas: at.AccountRAMDeltas,
+		}
+
+		if idx, exists := trxIndex[at.TrxID]; exists {
+			results[idx].Actions = append(results[idx].Actions, action)
+		} else {
+			trxIndex[at.TrxID] = len(results)
+			results = append(results, UsageTransaction{
+				TrxID:         at.TrxID,
+				BlockNum:      at.BlockNum,
+				BlockTime:     at.BlockTime,
+				CpuUsageUs:    at.CpuUsageUs,
+				NetUsageWords: at.NetUsageWords,
+				Actions:       []UsageAction{action},
+			})
+		}
+	}
+
+	return results, timings, nil
+}
diff --git a/services/actionindex/internal/stream_catchup.go b/services/actionindex/internal/stream_catchup.go
index 282e611..4f093ae 100644
--- a/services/actionindex/internal/stream_catchup.go
+++ b/services/actionindex/internal/stream_catchup.go
@@ -227,12 +227,14 @@ func (c *StreamCatchup) processBatch(seqs []uint64, sendAction func(StreamedActi
 	sent := 0
 	for j, at := range actions {
 		action := StreamedAction{
-			GlobalSeq: seqs[j],
-			BlockNum:  at.BlockNum,
-			BlockTime: chain.TimeToUint32(at.BlockTime),
-			Contract:  chain.StringToName(at.Act.Account),
-			Action:    chain.StringToName(at.Act.Name),
-			Receiver:  chain.StringToName(at.Receiver),
+			GlobalSeq:     seqs[j],
+			BlockNum:      at.BlockNum,
+			BlockTime:     chain.TimeToUint32(at.BlockTime),
+			Contract:      chain.StringToName(at.Act.Account),
+			Action:        chain.StringToName(at.Act.Name),
+			Receiver:      chain.StringToName(at.Receiver),
+			CpuUsageUs:    at.CpuUsageUs,
+			NetUsageWords: at.NetUsageWords,
 		}
 
 		if !c.filter.Matches(action) {
diff --git a/services/actionindex/internal/stream_tcp.go b/services/actionindex/internal/stream_tcp.go
index 46049cb..fa53f30 100644
--- a/services/actionindex/internal/stream_tcp.go
+++ b/services/actionindex/internal/stream_tcp.go
@@ -298,7 +298,7 @@ func (ts *StreamTCPServer) sendAction(conn net.Conn, action StreamedAction, deco
 		}
 	}
 
-	payload := make([]byte, 40+len(actionData))
+	payload := make([]byte, 48+len(actionData))
 
 	binary.LittleEndian.PutUint64(payload[0:8], action.GlobalSeq)
 	binary.LittleEndian.PutUint32(payload[8:12], action.BlockNum)
@@ -306,7 +306,9 @@ func (ts *StreamTCPServer) sendAction(conn net.Conn, action StreamedAction, deco
 	binary.LittleEndian.PutUint64(payload[16:24], action.Contract)
 	binary.LittleEndian.PutUint64(payload[24:32], action.Action)
 	binary.LittleEndian.PutUint64(payload[32:40], action.Receiver)
-	copy(payload[40:], actionData)
+	binary.LittleEndian.PutUint32(payload[40:44], action.CpuUsageUs)
+	binary.LittleEndian.PutUint32(payload[44:48], action.NetUsageWords)
+	copy(payload[48:], actionData)
 
 	return ts.writeMessage(conn, msgType, payload)
 }
diff --git a/services/actionindex/internal/sync.go b/services/actionindex/internal/sync.go
index fc63dd2..c4b7a4f 100644
--- a/services/actionindex/internal/sync.go
+++ b/services/actionindex/internal/sync.go
@@ -347,14 +347,17 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 		for _, idx := range matchingActions {
 			a := &block.Actions[idx]
 			actionData := block.GetActionDataBySeq(a.GlobalSeq)
+			cpuUs, netWords := block.GetResourceUsage(a.GlobalSeq)
 			if p.syncer.broadcaster.Broadcast(StreamedAction{
-				GlobalSeq:  a.GlobalSeq,
-				BlockNum:   block.BlockNum,
-				BlockTime:  block.BlockTime,
-				Contract:   a.Contract,
-				Action:     a.Action,
-				Receiver:   a.Account,
-				ActionData: actionData,
+				GlobalSeq:     a.GlobalSeq,
+				BlockNum:      block.BlockNum,
+				BlockTime:     block.BlockTime,
+				Contract:      a.Contract,
+				Action:        a.Action,
+				Receiver:      a.Account,
+				ActionData:    actionData,
+				CpuUsageUs:    cpuUs,
+				NetUsageWords: netWords,
 			}) {
 				delivered++
 			}
@@ -390,13 +393,15 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 				actionData, _ = hex.DecodeString(at.Act.Data)
 			}
 			if p.syncer.broadcaster.Broadcast(StreamedAction{
-				GlobalSeq:  seqs[i],
-				BlockNum:   block.BlockNum,
-				BlockTime:  block.BlockTime,
-				Contract:   block.Actions[idx].Contract,
-				Action:     block.Actions[idx].Action,
-				Receiver:   block.Actions[idx].Account,
-				ActionData: actionData,
+				GlobalSeq:     seqs[i],
+				BlockNum:      block.BlockNum,
+				BlockTime:     block.BlockTime,
+				Contract:      block.Actions[idx].Contract,
+				Action:        block.Actions[idx].Action,
+				Receiver:      block.Actions[idx].Account,
+				ActionData:    actionData,
+				CpuUsageUs:    at.CpuUsageUs,
+				NetUsageWords: at.NetUsageWords,
 			}) {
 				delivered++
 			}

From 9fc6bd2499614eb1927e59f1b10423af84b6a064 Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Wed, 11 Feb 2026 16:51:41 -0800
Subject: [PATCH 04/11] Adding /action endpoint for lookups by sequence

---
 services/actionindex/cmd/actionindex/main.go |   2 +
 services/actionindex/internal/openapi.go     |   1 +
 services/actionindex/internal/openapi.yaml   | 103 +++++++++++++++++++
 services/actionindex/internal/rpc_action.go  |  97 +++++++++++++++++
 4 files changed, 203 insertions(+)
 create mode 100644 services/actionindex/internal/rpc_action.go

diff --git a/services/actionindex/cmd/actionindex/main.go b/services/actionindex/cmd/actionindex/main.go
index ec66d7c..d2d774f 100644
--- a/services/actionindex/cmd/actionindex/main.go
+++ b/services/actionindex/cmd/actionindex/main.go
@@ -482,6 +482,8 @@ func handleRPC(
 		internal.HandleAccountLog(cfg, store, indexes, reader, abiReader, w, r)
 	case strings.HasPrefix(path, "/account/") && strings.HasSuffix(path, "/stats"):
 		internal.HandleAccountStats(indexes, w, r)
+	case strings.HasPrefix(path, "/action/"):
+		internal.HandleAction(cfg, reader, abiReader, w, r)
 	case strings.HasPrefix(path, "/debug/"):
 		if !cfg.DebugEndpoints {
 			http.Error(w, "debug endpoints not enabled (use --debug-endpoints)", http.StatusNotFound)
diff --git a/services/actionindex/internal/openapi.go b/services/actionindex/internal/openapi.go
index 0145635..9408673 100644
--- a/services/actionindex/internal/openapi.go
+++ b/services/actionindex/internal/openapi.go
@@ -30,6 +30,7 @@ var registeredRoutes = map[string][]string{
 	"/account/{account}/usage":    {"GET"},
 	"/account/{account}/log":      {"GET"},
 	"/account/{account}/stats":    {"GET"},
+	"/action/{seq}":               {"GET"},
 }
 
 var debugRoutes = map[string][]string{
diff --git a/services/actionindex/internal/openapi.yaml b/services/actionindex/internal/openapi.yaml
index a866f0d..6b1dde9 100644
--- a/services/actionindex/internal/openapi.yaml
+++ b/services/actionindex/internal/openapi.yaml
@@ -13,6 +13,8 @@ tags:
     description: Account activity and action history
   - name: Legacy Account History
     description: v1 API endpoints for account history
+  - name: Actions
+    description: Direct action lookup by global sequence number
   - name: Debug
     description: Internal state inspection (requires --debug-endpoints)
 
@@ -475,6 +477,61 @@ paths:
               schema:
                 $ref: "#/components/schemas/AccountStatsErrorResponse"
 
+  /action/{seq}:
+    get:
+      operationId: getActionBySeq
+      summary: Get action by global sequence
+      description: Look up a single action by its global sequence number. Returns either a minimal action object or a full action trace.
+      tags:
+        - Actions
+      parameters:
+        - name: seq
+          in: path
+          required: true
+          schema:
+            type: integer
+            format: uint64
+          description: Global sequence number
+        - name: trace
+          in: query
+          required: false
+          schema:
+            type: boolean
+            default: false
+          description: Return full action trace instead of minimal action
+        - name: decode
+          in: query
+          required: false
+          schema:
+            type: boolean
+            default: true
+          description: Decode action data using ABI
+      responses:
+        "200":
+          description: Action found
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ActionLookupResponse"
+        "400":
+          description: Invalid sequence number
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AccountActivityErrorResponse"
+        "404":
+          description: Action not found
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AccountActivityErrorResponse"
+        "500":
+          description: Internal error
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AccountActivityErrorResponse"
+
   /debug/account:
     get:
       operationId: debugAccount
@@ -882,6 +939,52 @@ components:
             message:
               type: string
 
+    ActionLookupResponse:
+      type: object
+      description: Single action lookup result
+      required:
+        - global_action_seq
+        - block_num
+        - block_time
+        - trx_id
+      properties:
+        global_action_seq:
+          type: integer
+          format: uint64
+        block_num:
+          type: integer
+          format: uint32
+        block_time:
+          type: string
+        action:
+          type: object
+          description: Minimal action object (when trace=false)
+          properties:
+            account:
+              type: string
+            name:
+              type: string
+            authorization:
+              type: array
+              items:
+                type: object
+                properties:
+                  actor:
+                    type: string
+                  permission:
+                    type: string
+            data:
+              oneOf:
+                - type: object
+                - type: string
+            hex_data:
+              type: string
+        action_trace:
+          $ref: "#/components/schemas/ActionTrace"
+          description: Full action trace (when trace=true)
+        trx_id:
+          type: string
+
     ActionResult:
       type: object
       description: Action result for legacy get_actions endpoint
diff --git a/services/actionindex/internal/rpc_action.go b/services/actionindex/internal/rpc_action.go
new file mode 100644
index 0000000..69fc4e3
--- /dev/null
+++ b/services/actionindex/internal/rpc_action.go
@@ -0,0 +1,97 @@
+package internal
+
+import (
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/greymass/roborovski/libraries/abicache"
+	"github.com/greymass/roborovski/libraries/corereader"
+	"github.com/greymass/roborovski/libraries/server"
+)
+
+type ActionLookupResponse struct {
+	GlobalActionSeq uint64      `json:"global_action_seq"`
+	BlockNum        uint32      `json:"block_num"`
+	BlockTime       string      `json:"block_time"`
+	Action          any `json:"action,omitempty"`
+	ActionTrace     any `json:"action_trace,omitempty"`
+	TrxID           string      `json:"trx_id"`
+}
+
+type ActionLookupError struct {
+	Code    string `json:"code"`
+	Message string `json:"message"`
+}
+
+type ActionLookupErrorResponse struct {
+	Error ActionLookupError `json:"error"`
+}
+
+func writeActionLookupError(w http.ResponseWriter, code, message string, status int) {
+	server.WriteJSON(w, status, ActionLookupErrorResponse{
+		Error: ActionLookupError{
+			Code:    code,
+			Message: message,
+		},
+	})
+}
+
+func HandleAction(cfg *Config, reader corereader.Reader, abiReader *abicache.Reader, w http.ResponseWriter, r *http.Request) {
+	seqStr := strings.TrimPrefix(r.URL.Path, "/action/")
+	if seqStr == "" || strings.Contains(seqStr, "/") {
+		writeActionLookupError(w, "missing_seq", "sequence number required in path", http.StatusBadRequest)
+		return
+	}
+
+	seq, err := strconv.ParseUint(seqStr, 10, 64)
+	if err != nil {
+		writeActionLookupError(w, "invalid_seq", "sequence must be a valid uint64", http.StatusBadRequest)
+		return
+	}
+
+	query := r.URL.Query()
+
+	trace := false
+	if v := query.Get("trace"); v == "true" || v == "1" {
+		trace = true
+	}
+
+	decode := true
+	if v := query.Get("decode"); v == "false" || v == "0" {
+		decode = false
+	}
+
+	actions, _, err := getActionsByGlobalSeqs(reader, []uint64{seq})
+	if err != nil {
+		writeActionLookupError(w, "internal_error", "failed to fetch action", http.StatusInternalServerError)
+		return
+	}
+
+	if len(actions) == 0 {
+		writeActionLookupError(w, "not_found", "action not found", http.StatusNotFound)
+		return
+	}
+
+	at := actions[0]
+
+	var effectiveAbiReader *abicache.Reader
+	if decode {
+		effectiveAbiReader = abiReader
+	}
+
+	resp := ActionLookupResponse{
+		GlobalActionSeq: seq,
+		BlockNum:        at.BlockNum,
+		BlockTime:       at.BlockTime,
+		TrxID:           at.TrxID,
+	}
+
+	if trace {
+		resp.ActionTrace = buildActionTrace(at, effectiveAbiReader, cfg.OmitNullFields)
+	} else {
+		resp.Action = buildAction(at.Act, effectiveAbiReader, at.BlockNum, cfg.OmitNullFields)
+	}
+
+	writeJSON(w, resp)
+}

From af4c8dcc3dddb7e83e73c5c95ad9a64359f08cff Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Thu, 5 Mar 2026 10:34:49 -0800
Subject: [PATCH 05/11] Add actions + seq types on websocket stream

---
 .../actionindex/internal/stream_websocket.go  | 19 ++++++++++++-------
 1 file changed, 12 insertions(+), 7 deletions(-)

diff --git a/services/actionindex/internal/stream_websocket.go b/services/actionindex/internal/stream_websocket.go
index eac92a0..d82326d 100644
--- a/services/actionindex/internal/stream_websocket.go
+++ b/services/actionindex/internal/stream_websocket.go
@@ -20,13 +20,14 @@ type wsSubscribeMessage struct {
 	Type      string   `json:"type"`
 	Contracts []string `json:"contracts,omitempty"`
 	Receivers []string `json:"receivers,omitempty"`
-	StartSeq  uint64   `json:"start_seq,omitempty"`
+	Actions   []string `json:"actions,omitempty"`
+	StartSeq  uint64   `json:"start_seq,string,omitempty"`
 	Decode    *bool    `json:"decode,omitempty"`
 }
 
 type wsActionMessage struct {
 	Type      string         `json:"type"`
-	GlobalSeq uint64         `json:"global_seq"`
+	GlobalSeq uint64         `json:"global_seq,string"`
 	BlockNum  uint32         `json:"block_num"`
 	BlockTime uint32         `json:"block_time"`
 	Contract  string         `json:"contract"`
@@ -38,14 +39,14 @@ type wsActionMessage struct {
 
 type wsHeartbeatMessage struct {
 	Type    string `json:"type"`
-	HeadSeq uint64 `json:"head_seq"`
-	LibSeq  uint64 `json:"lib_seq"`
+	HeadSeq uint64 `json:"head_seq,string"`
+	LibSeq  uint64 `json:"lib_seq,string"`
 }
 
 type wsCatchupCompleteMessage struct {
 	Type    string `json:"type"`
-	HeadSeq uint64 `json:"head_seq"`
-	LibSeq  uint64 `json:"lib_seq"`
+	HeadSeq uint64 `json:"head_seq,string"`
+	LibSeq  uint64 `json:"lib_seq,string"`
 }
 
 type wsErrorMessage struct {
@@ -155,6 +156,7 @@ func (ws *StreamWebSocketServer) handleWebSocket(w http.ResponseWriter, r *http.
 	filter := ActionFilter{
 		Contracts: make(map[uint64]struct{}),
 		Receivers: make(map[uint64]struct{}),
+		Actions:   make(map[uint64]struct{}),
 	}
 	for _, c := range subMsg.Contracts {
 		filter.Contracts[chain.StringToName(c)] = struct{}{}
@@ -162,6 +164,9 @@ func (ws *StreamWebSocketServer) handleWebSocket(w http.ResponseWriter, r *http.
 	for _, r := range subMsg.Receivers {
 		filter.Receivers[chain.StringToName(r)] = struct{}{}
 	}
+	for _, a := range subMsg.Actions {
+		filter.Actions[chain.StringToName(a)] = struct{}{}
+	}
 
 	decode := true
 	if subMsg.Decode != nil {
@@ -240,7 +245,7 @@ func (ws *StreamWebSocketServer) recvLoop(ctx context.Context, wsc *wsStreamClie
 		switch baseMsg.Type {
 		case "ack":
 			var ack struct {
-				Seq uint64 `json:"seq"`
+				Seq uint64 `json:"seq,string"`
 			}
 			if err := json.Unmarshal(msg, &ack); err == nil && wsc.client.sub != nil {
 				wsc.client.sub.Ack(ack.Seq)

From 0daef7741957e06180c12eb6592d5aa0cb476c4e Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Sat, 14 Mar 2026 20:35:11 -0700
Subject: [PATCH 06/11] Persist checkpoint if interrupted during bulk sync

---
 services/txindex/internal/bulksync.go | 35 +++++++++++++++------------
 1 file changed, 19 insertions(+), 16 deletions(-)

diff --git a/services/txindex/internal/bulksync.go b/services/txindex/internal/bulksync.go
index ecd15a4..ecd384b 100644
--- a/services/txindex/internal/bulksync.go
+++ b/services/txindex/internal/bulksync.go
@@ -27,16 +27,17 @@ type BulkIndexer struct {
 	lastBlock   uint32
 	logInterval time.Duration
 
-	activeBuffer  []WALEntry
-	flushBuffer   []WALEntry
-	activeBuffNum int
-	flushBuffNum  int
-	flushMu       sync.Mutex
-	flushCond     *sync.Cond
-	flushPending  bool
-	flushPhase    string
-	flushErr      error
-	flushDone     chan struct{}
+	activeBuffer     []WALEntry
+	flushBuffer      []WALEntry
+	activeBuffNum    int
+	flushBuffNum     int
+	flushLastBlock   uint32
+	flushMu          sync.Mutex
+	flushCond        *sync.Cond
+	flushPending     bool
+	flushPhase       string
+	flushErr         error
+	flushDone        chan struct{}
 
 	timing *BulkSyncTiming
 }
@@ -134,6 +135,7 @@ func (b *BulkIndexer) swapAndFlush() error {
 
 	b.activeBuffer, b.flushBuffer = b.flushBuffer, b.activeBuffer
 	b.flushBuffNum = b.activeBuffNum
+	b.flushLastBlock = b.lastBlock
 	b.activeBuffNum++
 	b.flushPending = true
 	b.flushCond.Signal()
@@ -193,16 +195,17 @@ func (b *BulkIndexer) doFlush(buffer []WALEntry) error {
 
 	b.runCount++
 
-	b.idx.SetBulkProgress(b.runCount, b.lastBlock)
-	if err := b.idx.SaveMeta(); err != nil {
-		logger.Printf("sync", "Warning: failed to save bulk progress: %v", err)
-	}
-
 	b.flushMu.Lock()
+	checkpointBlock := b.flushLastBlock
 	bufNum := b.flushBuffNum
 	b.flushMu.Unlock()
 
-	logger.Printf("sync", "Flushed buf %d with %s entries (checkpoint: block %s)", bufNum, logger.FormatCount(int64(len(buffer))), logger.FormatCount(int64(b.lastBlock)))
+	b.idx.SetBulkProgress(b.runCount, checkpointBlock)
+	if err := b.idx.SaveMeta(); err != nil {
+		logger.Printf("sync", "Warning: failed to save bulk progress: %v", err)
+	}
+
+	logger.Printf("sync", "Flushed buf %d with %s entries (checkpoint: block %s)", bufNum, logger.FormatCount(int64(len(buffer))), logger.FormatCount(int64(checkpointBlock)))
 	return nil
 }
 

From de3fccc2a5786ce915572fd2828fcc6d5cea8d24 Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Sun, 5 Apr 2026 16:46:36 -0700
Subject: [PATCH 07/11] Action Data retry with backoff

---
 services/actionindex/internal/sync.go         | 39 +++++++-
 .../actionindex/internal/sync_retry_test.go   | 97 +++++++++++++++++++
 2 files changed, 132 insertions(+), 4 deletions(-)
 create mode 100644 services/actionindex/internal/sync_retry_test.go

diff --git a/services/actionindex/internal/sync.go b/services/actionindex/internal/sync.go
index c4b7a4f..29ed015 100644
--- a/services/actionindex/internal/sync.go
+++ b/services/actionindex/internal/sync.go
@@ -7,6 +7,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/greymass/roborovski/libraries/chain"
 	"github.com/greymass/roborovski/libraries/corereader"
 	"github.com/greymass/roborovski/libraries/logger"
 )
@@ -374,11 +375,10 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 			seqs[i] = block.Actions[idx].GlobalSeq
 		}
 
-		traces, _, err := p.syncer.reader.GetActionsByGlobalSeqs(seqs)
+		traces, _, err := p.fetchActionDataWithRetry(block.BlockNum, seqs)
 		if err != nil {
-			errMsg := fmt.Sprintf("failed to fetch action data for block %d: %v", block.BlockNum, err)
-			p.syncer.broadcaster.BroadcastError(ActionErrorDataInconsistent, errMsg)
-			return fmt.Errorf("%s", errMsg)
+			p.syncer.broadcaster.BroadcastError(ActionErrorDataInconsistent, err.Error())
+			return err
 		}
 
 		var sendStart time.Time
@@ -421,6 +421,37 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 	return nil
 }
 
+func (p *AccountHistoryProcessor) fetchActionDataWithRetry(blockNum uint32, seqs []uint64) ([]chain.ActionTrace, *corereader.FetchTimings, error) {
+	const maxAttempts = 10
+	const initialDelay = 200 * time.Millisecond
+
+	delay := initialDelay
+	start := time.Now()
+
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
+		traces, timings, err := p.syncer.reader.GetActionsByGlobalSeqs(seqs)
+		if err == nil {
+			if attempt > 1 {
+				logger.Printf("warning", "Fetched action data for block %d after %d attempts (%v elapsed)",
+					blockNum, attempt, time.Since(start))
+			}
+			return traces, timings, nil
+		}
+
+		if attempt == maxAttempts {
+			return nil, nil, fmt.Errorf("failed to fetch action data for block %d after %d attempts (%v elapsed): %v",
+				blockNum, maxAttempts, time.Since(start), err)
+		}
+
+		logger.Printf("warning", "Failed to fetch action data for block %d (attempt %d/%d), retrying in %v: %v",
+			blockNum, attempt, maxAttempts, delay, err)
+		time.Sleep(delay)
+		delay *= 2
+	}
+
+	return nil, nil, nil
+}
+
 func (p *AccountHistoryProcessor) ShouldCommit(blocksProcessed int) bool {
 	return blocksProcessed >= int(p.syncer.bulkCommitInterval) || blocksProcessed < 50
 }
diff --git a/services/actionindex/internal/sync_retry_test.go b/services/actionindex/internal/sync_retry_test.go
new file mode 100644
index 0000000..e2f3be7
--- /dev/null
+++ b/services/actionindex/internal/sync_retry_test.go
@@ -0,0 +1,97 @@
+package internal
+
+import (
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/greymass/roborovski/libraries/chain"
+	"github.com/greymass/roborovski/libraries/corereader"
+)
+
+type mockReader struct {
+	corereader.Reader
+	callCount  atomic.Int32
+	failUntil  int32
+	returnData []chain.ActionTrace
+}
+
+func (m *mockReader) GetActionsByGlobalSeqs(globalSeqs []uint64) ([]chain.ActionTrace, *corereader.FetchTimings, error) {
+	n := m.callCount.Add(1)
+	if n <= m.failUntil {
+		return nil, nil, fmt.Errorf("glob %d not found in any slice", globalSeqs[0])
+	}
+	return m.returnData, &corereader.FetchTimings{}, nil
+}
+
+func TestFetchActionDataWithRetry_SucceedsOnFirstAttempt(t *testing.T) {
+	mock := &mockReader{
+		failUntil:  0,
+		returnData: []chain.ActionTrace{{Receiver: "test"}},
+	}
+	p := &AccountHistoryProcessor{
+		syncer: &Syncer{reader: mock},
+	}
+
+	traces, _, err := p.fetchActionDataWithRetry(100, []uint64{5000})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(traces) != 1 || traces[0].Receiver != "test" {
+		t.Fatalf("unexpected traces: %v", traces)
+	}
+	if mock.callCount.Load() != 1 {
+		t.Errorf("expected 1 call, got %d", mock.callCount.Load())
+	}
+}
+
+func TestFetchActionDataWithRetry_SucceedsAfterRetries(t *testing.T) {
+	mock := &mockReader{
+		failUntil:  3,
+		returnData: []chain.ActionTrace{{Receiver: "recovered"}},
+	}
+	p := &AccountHistoryProcessor{
+		syncer: &Syncer{reader: mock},
+	}
+
+	start := time.Now()
+	traces, _, err := p.fetchActionDataWithRetry(100, []uint64{5000})
+	elapsed := time.Since(start)
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(traces) != 1 || traces[0].Receiver != "recovered" {
+		t.Fatalf("unexpected traces: %v", traces)
+	}
+	if mock.callCount.Load() != 4 {
+		t.Errorf("expected 4 calls (3 failures + 1 success), got %d", mock.callCount.Load())
+	}
+	if elapsed < 200*time.Millisecond+400*time.Millisecond+800*time.Millisecond {
+		t.Errorf("expected at least 1.4s of backoff delay, completed in %v", elapsed)
+	}
+	t.Logf("Succeeded on attempt 4 after %v", elapsed)
+}
+
+func TestFetchActionDataWithRetry_FailsAfterMaxAttempts(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping long retry test in short mode")
+	}
+
+	mock := &mockReader{
+		failUntil: 100,
+	}
+	p := &AccountHistoryProcessor{
+		syncer: &Syncer{reader: mock},
+	}
+
+	_, _, err := p.fetchActionDataWithRetry(100, []uint64{5000})
+	if err == nil {
+		t.Fatal("expected error after max retries, got nil")
+	}
+	if mock.callCount.Load() != 10 {
+		t.Errorf("expected 10 calls, got %d", mock.callCount.Load())
+	}
+	t.Logf("Failed as expected after %d attempts: %v", mock.callCount.Load(), err)
+}

From 5b334d2b57084f9139fafdb314e492c585f536be Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Thu, 23 Apr 2026 19:00:23 -0700
Subject: [PATCH 08/11] Fixed authorizer-indexed actions being dropped in
 stream catchup

---
 .../actionindex/internal/stream_catchup.go    |  29 ++-
 .../internal/stream_catchup_test.go           | 201 ++++++++++++++++++
 2 files changed, 222 insertions(+), 8 deletions(-)
 create mode 100644 services/actionindex/internal/stream_catchup_test.go

diff --git a/services/actionindex/internal/stream_catchup.go b/services/actionindex/internal/stream_catchup.go
index 4f093ae..6f1c8b9 100644
--- a/services/actionindex/internal/stream_catchup.go
+++ b/services/actionindex/internal/stream_catchup.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"context"
 	"encoding/hex"
+	"slices"
 
 	"github.com/greymass/roborovski/libraries/chain"
 	"github.com/greymass/roborovski/libraries/corereader"
@@ -113,7 +114,7 @@ func (c *StreamCatchup) streamWithContractActionIndex(ctx context.Context, sendA
 			end = len(allSeqs)
 		}
 
-		sent, err := c.processBatch(allSeqs[i:end], sendAction)
+		sent, err := c.processBatch(allSeqs[i:end], contract, nil, sendAction)
 		if err != nil {
 			return err
 		}
@@ -139,7 +140,7 @@ func (c *StreamCatchup) streamSingleAccount(ctx context.Context, account uint64,
 		batch = append(batch, seq)
 
 		if len(batch) >= StreamCatchupBatchSize {
-			sent, err := c.processBatch(batch, sendAction)
+			sent, err := c.processBatch(batch, account, nil, sendAction)
 			if err != nil {
 				return err
 			}
@@ -155,7 +156,7 @@ func (c *StreamCatchup) streamSingleAccount(ctx context.Context, account uint64,
 
 	// Process remaining batch
 	if len(batch) > 0 {
-		sent, err := c.processBatch(batch, sendAction)
+		sent, err := c.processBatch(batch, account, nil, sendAction)
 		if err != nil {
 			return err
 		}
@@ -167,13 +168,19 @@ func (c *StreamCatchup) streamSingleAccount(ctx context.Context, account uint64,
 }
 
 func (c *StreamCatchup) streamMultipleAccounts(ctx context.Context, accounts []uint64, sendAction func(StreamedAction) error) error {
+	seqToAccount := make(map[uint64]uint64)
 	var allSeqs []uint64
 	for _, account := range accounts {
 		seqs, err := c.indexes.chunkReader.GetWithSeqRange(account, c.startSeq, c.endSeq, unlimitedLimit, false)
 		if err != nil {
 			return err
 		}
-		allSeqs = mergeAndDedupe(allSeqs, seqs)
+		for _, seq := range seqs {
+			if _, ok := seqToAccount[seq]; !ok {
+				seqToAccount[seq] = account
+				allSeqs = append(allSeqs, seq)
+			}
+		}
 	}
 
 	if len(allSeqs) == 0 {
@@ -181,6 +188,8 @@ func (c *StreamCatchup) streamMultipleAccounts(ctx context.Context, accounts []u
 		return nil
 	}
 
+	slices.Sort(allSeqs)
+
 	logger.Printf("stream", "Catchup: %d merged sequences from %d accounts", len(allSeqs), len(accounts))
 
 	var totalSent int
@@ -196,7 +205,7 @@ func (c *StreamCatchup) streamMultipleAccounts(ctx context.Context, accounts []u
 			end = len(allSeqs)
 		}
 
-		sent, err := c.processBatch(allSeqs[i:end], sendAction)
+		sent, err := c.processBatch(allSeqs[i:end], 0, seqToAccount, sendAction)
 		if err != nil {
 			return err
 		}
@@ -207,7 +216,7 @@ func (c *StreamCatchup) streamMultipleAccounts(ctx context.Context, accounts []u
 	return nil
 }
 
-func (c *StreamCatchup) processBatch(seqs []uint64, sendAction func(StreamedAction) error) (int, error) {
+func (c *StreamCatchup) processBatch(seqs []uint64, account uint64, seqToAccount map[uint64]uint64, sendAction func(StreamedAction) error) (int, error) {
 	actions, _, err := c.reader.GetActionsByGlobalSeqs(seqs)
 	if err != nil {
 		if len(seqs) == 1 {
@@ -215,7 +224,7 @@ func (c *StreamCatchup) processBatch(seqs []uint64, sendAction func(StreamedActi
 		}
 		sent := 0
 		for _, seq := range seqs {
-			n, err := c.processBatch([]uint64{seq}, sendAction)
+			n, err := c.processBatch([]uint64{seq}, account, seqToAccount, sendAction)
 			if err != nil {
 				return sent, err
 			}
@@ -226,13 +235,17 @@ func (c *StreamCatchup) processBatch(seqs []uint64, sendAction func(StreamedActi
 
 	sent := 0
 	for j, at := range actions {
+		receiver := account
+		if seqToAccount != nil {
+			receiver = seqToAccount[seqs[j]]
+		}
 		action := StreamedAction{
 			GlobalSeq:     seqs[j],
 			BlockNum:      at.BlockNum,
 			BlockTime:     chain.TimeToUint32(at.BlockTime),
 			Contract:      chain.StringToName(at.Act.Account),
 			Action:        chain.StringToName(at.Act.Name),
-			Receiver:      chain.StringToName(at.Receiver),
+			Receiver:      receiver,
 			CpuUsageUs:    at.CpuUsageUs,
 			NetUsageWords: at.NetUsageWords,
 		}
diff --git a/services/actionindex/internal/stream_catchup_test.go b/services/actionindex/internal/stream_catchup_test.go
new file mode 100644
index 0000000..9cdaf96
--- /dev/null
+++ b/services/actionindex/internal/stream_catchup_test.go
@@ -0,0 +1,201 @@
+package internal
+
+import (
+	"testing"
+
+	"github.com/greymass/roborovski/libraries/chain"
+	"github.com/greymass/roborovski/libraries/corereader"
+)
+
+type stubBaseReader struct {
+	corereader.BaseReader
+	traces   []chain.ActionTrace
+	bySeq    map[uint64]chain.ActionTrace
+}
+
+func (s *stubBaseReader) GetActionsByGlobalSeqs(seqs []uint64) ([]chain.ActionTrace, *corereader.FetchTimings, error) {
+	if s.bySeq != nil {
+		out := make([]chain.ActionTrace, 0, len(seqs))
+		for _, seq := range seqs {
+			out = append(out, s.bySeq[seq])
+		}
+		return out, &corereader.FetchTimings{}, nil
+	}
+	return s.traces, &corereader.FetchTimings{}, nil
+}
+
+func TestProcessBatch_DeliversAuthorizerOnlyAction(t *testing.T) {
+	tracked := chain.StringToName("shipload.gm")
+
+	setabiTrace := chain.ActionTrace{
+		Receiver: "eosio",
+		Act: chain.Action{
+			Account: "eosio",
+			Name:    "setabi",
+			Authorization: []chain.PermissionLevel{
+				{Actor: "shipload.gm", Permission: "active"},
+			},
+		},
+	}
+
+	catchup := &StreamCatchup{
+		reader: &stubBaseReader{traces: []chain.ActionTrace{setabiTrace}},
+		filter: ActionFilter{
+			Receivers: map[uint64]struct{}{tracked: {}},
+		},
+	}
+
+	var delivered []StreamedAction
+	sent, err := catchup.processBatch([]uint64{1000}, tracked, nil, func(a StreamedAction) error {
+		delivered = append(delivered, a)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("processBatch error: %v", err)
+	}
+	if sent != 1 {
+		t.Fatalf("expected 1 action delivered, got %d", sent)
+	}
+	if len(delivered) != 1 {
+		t.Fatalf("expected delivered len 1, got %d", len(delivered))
+	}
+	got := delivered[0]
+	if got.Contract != chain.StringToName("eosio") {
+		t.Errorf("Contract = %s, want eosio", chain.NameToString(got.Contract))
+	}
+	if got.Action != chain.StringToName("setabi") {
+		t.Errorf("Action = %s, want setabi", chain.NameToString(got.Action))
+	}
+	if got.Receiver != tracked {
+		t.Errorf("Receiver = %s, want shipload.gm (tracked account)", chain.NameToString(got.Receiver))
+	}
+}
+
+func TestProcessBatch_DeliversReceiverPathAction(t *testing.T) {
+	tracked := chain.StringToName("shipload.gm")
+
+	transferTrace := chain.ActionTrace{
+		Receiver: "shipload.gm",
+		Act: chain.Action{
+			Account: "eosio.token",
+			Name:    "transfer",
+		},
+	}
+
+	catchup := &StreamCatchup{
+		reader: &stubBaseReader{traces: []chain.ActionTrace{transferTrace}},
+		filter: ActionFilter{
+			Receivers: map[uint64]struct{}{tracked: {}},
+		},
+	}
+
+	var delivered []StreamedAction
+	sent, err := catchup.processBatch([]uint64{2000}, tracked, nil, func(a StreamedAction) error {
+		delivered = append(delivered, a)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("processBatch error: %v", err)
+	}
+	if sent != 1 || len(delivered) != 1 {
+		t.Fatalf("expected 1 delivery, got sent=%d delivered=%d", sent, len(delivered))
+	}
+	got := delivered[0]
+	if got.Contract != chain.StringToName("eosio.token") {
+		t.Errorf("Contract = %s, want eosio.token", chain.NameToString(got.Contract))
+	}
+	if got.Action != chain.StringToName("transfer") {
+		t.Errorf("Action = %s, want transfer", chain.NameToString(got.Action))
+	}
+	if got.Receiver != tracked {
+		t.Errorf("Receiver = %s, want shipload.gm", chain.NameToString(got.Receiver))
+	}
+}
+
+func TestProcessBatch_MultiAccount_AttributesPerSeq(t *testing.T) {
+	shipload := chain.StringToName("shipload.gm")
+	platform := chain.StringToName("platform.gm")
+
+	reader := &stubBaseReader{
+		bySeq: map[uint64]chain.ActionTrace{
+			100: {
+				Receiver: "eosio",
+				Act:      chain.Action{Account: "eosio", Name: "setabi"},
+			},
+			200: {
+				Receiver: "shipload.gm",
+				Act:      chain.Action{Account: "eosio.token", Name: "transfer"},
+			},
+			300: {
+				Receiver: "eosio",
+				Act:      chain.Action{Account: "eosio", Name: "setcode"},
+			},
+		},
+	}
+
+	seqToAccount := map[uint64]uint64{
+		100: platform,
+		200: shipload,
+		300: platform,
+	}
+
+	catchup := &StreamCatchup{
+		reader: reader,
+		filter: ActionFilter{
+			Receivers: map[uint64]struct{}{shipload: {}, platform: {}},
+		},
+	}
+
+	delivered := map[uint64]StreamedAction{}
+	sent, err := catchup.processBatch([]uint64{100, 200, 300}, 0, seqToAccount, func(a StreamedAction) error {
+		delivered[a.GlobalSeq] = a
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("processBatch error: %v", err)
+	}
+	if sent != 3 || len(delivered) != 3 {
+		t.Fatalf("expected 3 deliveries, got sent=%d delivered=%d", sent, len(delivered))
+	}
+	if got := delivered[100].Receiver; got != platform {
+		t.Errorf("seq 100 Receiver = %s, want platform.gm", chain.NameToString(got))
+	}
+	if got := delivered[200].Receiver; got != shipload {
+		t.Errorf("seq 200 Receiver = %s, want shipload.gm", chain.NameToString(got))
+	}
+	if got := delivered[300].Receiver; got != platform {
+		t.Errorf("seq 300 Receiver = %s, want platform.gm", chain.NameToString(got))
+	}
+}
+
+func TestProcessBatch_AppliesContractActionFilter(t *testing.T) {
+	tracked := chain.StringToName("shipload.gm")
+
+	transferTrace := chain.ActionTrace{
+		Receiver: "shipload.gm",
+		Act: chain.Action{
+			Account: "eosio.token",
+			Name:    "transfer",
+		},
+	}
+
+	catchup := &StreamCatchup{
+		reader: &stubBaseReader{traces: []chain.ActionTrace{transferTrace}},
+		filter: ActionFilter{
+			Contracts: map[uint64]struct{}{chain.StringToName("eosio"): {}},
+			Actions:   map[uint64]struct{}{chain.StringToName("setabi"): {}},
+			Receivers: map[uint64]struct{}{tracked: {}},
+		},
+	}
+
+	sent, err := catchup.processBatch([]uint64{1000}, tracked, nil, func(a StreamedAction) error {
+		t.Fatalf("did not expect delivery: %+v", a)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("processBatch error: %v", err)
+	}
+	if sent != 0 {
+		t.Fatalf("expected 0 deliveries, got %d", sent)
+	}
+}

From ee4955a7806fc9eaa8d56688fe61914619515c70 Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Mon, 18 May 2026 15:44:05 -0700
Subject: [PATCH 09/11] Match authorizing accounts without mutating receivers

---
 libraries/corereader/canonical_filter.go      |   4 +
 libraries/corereader/canonical_filter_test.go | 177 ++++++++++
 libraries/corereader/types.go                 |   8 +
 services/actionindex/internal/broadcaster.go  |  15 +-
 .../actionindex/internal/stream_catchup.go    |   9 +-
 .../internal/stream_catchup_test.go           |  74 ++++-
 services/actionindex/internal/stream_types.go |  15 +-
 .../actionindex/internal/stream_types_test.go |  77 ++++-
 services/actionindex/internal/sync.go         | 107 +++---
 .../internal/sync_broadcast_test.go           | 304 ++++++++++++++++++
 10 files changed, 705 insertions(+), 85 deletions(-)
 create mode 100644 services/actionindex/internal/sync_broadcast_test.go

diff --git a/libraries/corereader/canonical_filter.go b/libraries/corereader/canonical_filter.go
index 7632ca6..2afa45e 100644
--- a/libraries/corereader/canonical_filter.go
+++ b/libraries/corereader/canonical_filter.go
@@ -322,6 +322,7 @@ func filterBlockInto(bf *blockFilter, notif RawBlock, actionFilter ActionFilterF
 					GlobalSeq:    globalSeq,
 					TrxIndex:     info.TrxIndex,
 					IsAuthorizer: isAuth,
+					Receiver:     info.Receiver,
 				})
 				if minSeq == 0 || globalSeq < minSeq {
 					minSeq = globalSeq
@@ -377,6 +378,7 @@ func filterBlockInto(bf *blockFilter, notif RawBlock, actionFilter ActionFilterF
 				GlobalSeq:    globalSeq,
 				TrxIndex:     info.TrxIndex,
 				IsAuthorizer: true,
+				Receiver:     info.Receiver,
 			})
 			if minSeq == 0 || globalSeq < minSeq {
 				minSeq = globalSeq
@@ -504,6 +506,7 @@ func filterBlockIntoTimed(bf *blockFilter, notif RawBlock, actionFilter ActionFi
 					GlobalSeq:    globalSeq,
 					TrxIndex:     info.TrxIndex,
 					IsAuthorizer: isAuth,
+					Receiver:     info.Receiver,
 				})
 				if minSeq == 0 || globalSeq < minSeq {
 					minSeq = globalSeq
@@ -562,6 +565,7 @@ func filterBlockIntoTimed(bf *blockFilter, notif RawBlock, actionFilter ActionFi
 				GlobalSeq:    globalSeq,
 				TrxIndex:     info.TrxIndex,
 				IsAuthorizer: true,
+				Receiver:     info.Receiver,
 			})
 			if minSeq == 0 || globalSeq < minSeq {
 				minSeq = globalSeq
diff --git a/libraries/corereader/canonical_filter_test.go b/libraries/corereader/canonical_filter_test.go
index 98a9abc..8f9735f 100644
--- a/libraries/corereader/canonical_filter_test.go
+++ b/libraries/corereader/canonical_filter_test.go
@@ -309,3 +309,180 @@ func abs(x float64) float64 {
 	}
 	return x
 }
+
+func TestFilterRawBlock_ActionReceiver_ReceiverPath(t *testing.T) {
+	const (
+		acctIdx  = uint32(0)
+		contract = uint64(0xC0FFEE)
+		action   = uint64(0xACEACE)
+		seq      = uint64(42)
+		account  = uint64(0xA11CE)
+	)
+
+	raw := RawBlock{
+		BlockNum:     1,
+		BlockTime:    1000,
+		NamesInBlock: []uint64{account},
+		Notifications: map[uint64][]uint64{
+			account: {seq},
+		},
+		ActionMeta: []ActionMetadata{
+			{GlobalSeq: seq, Contract: contract, Action: action},
+		},
+		Actions: []CanonicalAction{
+			{
+				ActionOrdinal:      1,
+				CreatorAO:          0,
+				ReceiverUint64:     account, // chain receiver == account
+				DataIndex:          0,
+				AuthAccountIndexes: []uint32{acctIdx},
+				GlobalSeqUint64:    seq,
+				TrxIndex:           0,
+				ContractUint64:     contract,
+				ActionUint64:       action,
+			},
+		},
+	}
+
+	block := FilterRawBlock(raw, nil)
+	if len(block.Actions) != 1 {
+		t.Fatalf("expected 1 action, got %d", len(block.Actions))
+	}
+	got := block.Actions[0]
+	if got.Account != account {
+		t.Errorf("Account = %#x, want %#x", got.Account, account)
+	}
+	if got.Receiver != account {
+		t.Errorf("Receiver = %#x, want %#x (chain receiver == account on receiver path)", got.Receiver, account)
+	}
+}
+
+func TestFilterRawBlock_ActionReceiver_AuthorizerPath(t *testing.T) {
+	const (
+		contract    = uint64(0xC0FFEE)
+		action      = uint64(0xACEACE)
+		seq         = uint64(43)
+		chainRecv   = uint64(0xA11CE)
+		authAccount = uint64(0xB0B)
+	)
+
+	raw := RawBlock{
+		BlockNum:     1,
+		BlockTime:    1000,
+		NamesInBlock: []uint64{authAccount}, // index 0
+		Notifications: map[uint64][]uint64{
+			authAccount: {seq}, // authAccount tracked, gets notified for seq it didn't receive
+		},
+		ActionMeta: []ActionMetadata{
+			{GlobalSeq: seq, Contract: contract, Action: action},
+		},
+		Actions: []CanonicalAction{
+			{
+				ActionOrdinal:      1,
+				CreatorAO:          0,
+				ReceiverUint64:     chainRecv, // chain receiver != tracked account
+				DataIndex:          0,
+				AuthAccountIndexes: []uint32{0}, // index into NamesInBlock = authAccount
+				GlobalSeqUint64:    seq,
+				TrxIndex:           0,
+				ContractUint64:     contract,
+				ActionUint64:       action,
+			},
+		},
+	}
+
+	block := FilterRawBlock(raw, nil)
+	if len(block.Actions) != 1 {
+		t.Fatalf("expected 1 action, got %d", len(block.Actions))
+	}
+	got := block.Actions[0]
+	if got.Account != authAccount {
+		t.Errorf("Account = %#x, want %#x (authorizer-path Account = the authorizer)", got.Account, authAccount)
+	}
+	if got.Receiver != chainRecv {
+		t.Errorf("Receiver = %#x, want %#x (chain receiver of the underlying action_trace)", got.Receiver, chainRecv)
+	}
+	if !got.IsAuthorizer {
+		t.Errorf("IsAuthorizer = false, want true on authorizer-path emission")
+	}
+}
+
+func TestFilterRawBlockIntoTimed_ActionReceiver_BothPaths(t *testing.T) {
+	const (
+		contract    = uint64(0xC0FFEE)
+		action      = uint64(0xACEACE)
+		recvSeq     = uint64(50)
+		authSeq     = uint64(51)
+		recvAccount = uint64(0xA11CE)
+		chainRecv   = uint64(0xCA1F)
+		authAccount = uint64(0xB0B)
+	)
+
+	raw := RawBlock{
+		BlockNum:     1,
+		BlockTime:    1000,
+		NamesInBlock: []uint64{authAccount}, // index 0
+		Notifications: map[uint64][]uint64{
+			recvAccount: {recvSeq}, // receiver-path
+			authAccount: {authSeq}, // authorizer-path
+		},
+		ActionMeta: []ActionMetadata{
+			{GlobalSeq: recvSeq, Contract: contract, Action: action},
+			{GlobalSeq: authSeq, Contract: contract, Action: action},
+		},
+		Actions: []CanonicalAction{
+			{
+				ActionOrdinal:      1,
+				CreatorAO:          0,
+				ReceiverUint64:     recvAccount, // chain receiver == recvAccount on this seq
+				DataIndex:          0,
+				AuthAccountIndexes: []uint32{0}, // authAccount (incidentally also auth here, not tested)
+				GlobalSeqUint64:    recvSeq,
+				TrxIndex:           0,
+				ContractUint64:     contract,
+				ActionUint64:       action,
+			},
+			{
+				ActionOrdinal:      2,
+				CreatorAO:          0,
+				ReceiverUint64:     chainRecv, // chain receiver != authAccount on this seq
+				DataIndex:          0,
+				AuthAccountIndexes: []uint32{0}, // authAccount
+				GlobalSeqUint64:    authSeq,
+				TrxIndex:           0,
+				ContractUint64:     contract,
+				ActionUint64:       action,
+			},
+		},
+	}
+
+	timing := &FilterTiming{}
+	block, _, _ := FilterRawBlockIntoTimed(raw, nil, nil, nil, timing)
+
+	bySeq := map[uint64]Action{}
+	for _, a := range block.Actions {
+		bySeq[a.GlobalSeq] = a
+	}
+
+	recvEntry, ok := bySeq[recvSeq]
+	if !ok {
+		t.Fatalf("missing receiver-path entry for seq %d", recvSeq)
+	}
+	if recvEntry.Receiver != recvAccount {
+		t.Errorf("receiver-path Receiver = %#x, want %#x", recvEntry.Receiver, recvAccount)
+	}
+
+	authEntry, ok := bySeq[authSeq]
+	if !ok {
+		t.Fatalf("missing authorizer-path entry for seq %d", authSeq)
+	}
+	if authEntry.Account != authAccount {
+		t.Errorf("authorizer-path Account = %#x, want %#x", authEntry.Account, authAccount)
+	}
+	if authEntry.Receiver != chainRecv {
+		t.Errorf("authorizer-path Receiver = %#x, want %#x (chain receiver)", authEntry.Receiver, chainRecv)
+	}
+	if !authEntry.IsAuthorizer {
+		t.Errorf("authorizer-path IsAuthorizer = false, want true")
+	}
+}
diff --git a/libraries/corereader/types.go b/libraries/corereader/types.go
index d67d675..ae98e70 100644
--- a/libraries/corereader/types.go
+++ b/libraries/corereader/types.go
@@ -56,6 +56,13 @@ func (r *RawBlock) HasActionData() bool {
 	return r.rawData != nil
 }
 
+// SetActionData populates the raw action-data buffers (test helper).
+func (r *RawBlock) SetActionData(rawData []byte, offsets, lengths []uint32) {
+	r.rawData = rawData
+	r.dataOffsets = offsets
+	r.dataLengths = lengths
+}
+
 // Action represents a canonically-deduplicated action for a specific account.
 // This is the output of canonical filtering - one Action per (account, globalSeq) pair.
 type Action struct {
@@ -65,6 +72,7 @@ type Action struct {
 	GlobalSeq    uint64
 	TrxIndex     uint32
 	IsAuthorizer bool
+	Receiver     uint64 // chain receiver; differs from Account on authorizer-indexed entries
 }
 
 // ContractExecution represents a contract execution (receiver == contract).
diff --git a/services/actionindex/internal/broadcaster.go b/services/actionindex/internal/broadcaster.go
index 9c4efe3..c081d42 100644
--- a/services/actionindex/internal/broadcaster.go
+++ b/services/actionindex/internal/broadcaster.go
@@ -74,7 +74,7 @@ func (b *ActionBroadcaster) Unsubscribe(id uint64) {
 	}
 }
 
-func (b *ActionBroadcaster) Broadcast(action StreamedAction) bool {
+func (b *ActionBroadcaster) Broadcast(action StreamedAction, matchedVia []uint64) bool {
 	if b.closed.Load() || !b.liveMode.Load() {
 		return false
 	}
@@ -87,9 +87,9 @@ func (b *ActionBroadcaster) Broadcast(action StreamedAction) bool {
 		if sub.IsCatchingUp() {
 			continue
 		}
-		if !sub.filter.Matches(action) {
-			logger.Printf("debug-stream", "Subscription %d filter rejected: contract=%x receiver=%x action=%x (filter: contracts=%d receivers=%d actions=%d)",
-				sub.id, action.Contract, action.Receiver, action.Action,
+		if !sub.filter.Matches(action, matchedVia) {
+			logger.Printf("debug-stream", "Subscription %d filter rejected: contract=%x receiver=%x action=%x matchedVia=%v (filter: contracts=%d receivers=%d actions=%d)",
+				sub.id, action.Contract, action.Receiver, action.Action, matchedVia,
 				len(sub.filter.Contracts), len(sub.filter.Receivers), len(sub.filter.Actions))
 			continue
 		}
@@ -163,7 +163,7 @@ func (b *ActionBroadcaster) SubscriberCount() int {
 	return len(b.subs)
 }
 
-func (b *ActionBroadcaster) CouldMatch(contract, action, receiver uint64) bool {
+func (b *ActionBroadcaster) CouldMatch(contract, action uint64, matchedVia []uint64) bool {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
@@ -189,13 +189,14 @@ func (b *ActionBroadcaster) CouldMatch(contract, action, receiver uint64) bool {
 		}
 
 		if len(f.Receivers) == 0 {
-			if contractMatch && receiver == contract {
+			if contractMatch {
 				return true
 			}
 			continue
 		}
 
-		_, receiverMatch := f.Receivers[receiver]
+		receiverMatch := anyInSet(matchedVia, f.Receivers)
+
 		if len(f.Contracts) == 0 {
 			if receiverMatch {
 				return true
diff --git a/services/actionindex/internal/stream_catchup.go b/services/actionindex/internal/stream_catchup.go
index 6f1c8b9..cf4ae3d 100644
--- a/services/actionindex/internal/stream_catchup.go
+++ b/services/actionindex/internal/stream_catchup.go
@@ -234,10 +234,11 @@ func (c *StreamCatchup) processBatch(seqs []uint64, account uint64, seqToAccount
 	}
 
 	sent := 0
+	var matchedViaBuf [1]uint64
 	for j, at := range actions {
-		receiver := account
+		matchedViaBuf[0] = account
 		if seqToAccount != nil {
-			receiver = seqToAccount[seqs[j]]
+			matchedViaBuf[0] = seqToAccount[seqs[j]]
 		}
 		action := StreamedAction{
 			GlobalSeq:     seqs[j],
@@ -245,12 +246,12 @@ func (c *StreamCatchup) processBatch(seqs []uint64, account uint64, seqToAccount
 			BlockTime:     chain.TimeToUint32(at.BlockTime),
 			Contract:      chain.StringToName(at.Act.Account),
 			Action:        chain.StringToName(at.Act.Name),
-			Receiver:      receiver,
+			Receiver:      chain.StringToName(at.Receiver),
 			CpuUsageUs:    at.CpuUsageUs,
 			NetUsageWords: at.NetUsageWords,
 		}
 
-		if !c.filter.Matches(action) {
+		if !c.filter.Matches(action, matchedViaBuf[:]) {
 			continue
 		}
 
diff --git a/services/actionindex/internal/stream_catchup_test.go b/services/actionindex/internal/stream_catchup_test.go
index 9cdaf96..c32b9e0 100644
--- a/services/actionindex/internal/stream_catchup_test.go
+++ b/services/actionindex/internal/stream_catchup_test.go
@@ -66,8 +66,9 @@ func TestProcessBatch_DeliversAuthorizerOnlyAction(t *testing.T) {
 	if got.Action != chain.StringToName("setabi") {
 		t.Errorf("Action = %s, want setabi", chain.NameToString(got.Action))
 	}
-	if got.Receiver != tracked {
-		t.Errorf("Receiver = %s, want shipload.gm (tracked account)", chain.NameToString(got.Receiver))
+	if got.Receiver != chain.StringToName(setabiTrace.Receiver) {
+		t.Errorf("Receiver = %s, want %s (chain receiver, not indexed account)",
+			chain.NameToString(got.Receiver), setabiTrace.Receiver)
 	}
 }
 
@@ -107,8 +108,9 @@ func TestProcessBatch_DeliversReceiverPathAction(t *testing.T) {
 	if got.Action != chain.StringToName("transfer") {
 		t.Errorf("Action = %s, want transfer", chain.NameToString(got.Action))
 	}
-	if got.Receiver != tracked {
-		t.Errorf("Receiver = %s, want shipload.gm", chain.NameToString(got.Receiver))
+	if got.Receiver != chain.StringToName(transferTrace.Receiver) {
+		t.Errorf("Receiver = %s, want %s (chain receiver)",
+			chain.NameToString(got.Receiver), transferTrace.Receiver)
 	}
 }
 
@@ -157,14 +159,12 @@ func TestProcessBatch_MultiAccount_AttributesPerSeq(t *testing.T) {
 	if sent != 3 || len(delivered) != 3 {
 		t.Fatalf("expected 3 deliveries, got sent=%d delivered=%d", sent, len(delivered))
 	}
-	if got := delivered[100].Receiver; got != platform {
-		t.Errorf("seq 100 Receiver = %s, want platform.gm", chain.NameToString(got))
-	}
-	if got := delivered[200].Receiver; got != shipload {
-		t.Errorf("seq 200 Receiver = %s, want shipload.gm", chain.NameToString(got))
-	}
-	if got := delivered[300].Receiver; got != platform {
-		t.Errorf("seq 300 Receiver = %s, want platform.gm", chain.NameToString(got))
+	for _, seq := range []uint64{100, 200, 300} {
+		want := chain.StringToName(reader.bySeq[seq].Receiver)
+		if got := delivered[seq].Receiver; got != want {
+			t.Errorf("seq %d Receiver = %s, want %s (chain receiver from trace)",
+				seq, chain.NameToString(got), chain.NameToString(want))
+		}
 	}
 }
 
@@ -199,3 +199,53 @@ func TestProcessBatch_AppliesContractActionFilter(t *testing.T) {
 		t.Fatalf("expected 0 deliveries, got %d", sent)
 	}
 }
+
+func TestProcessBatch_AuthorizerOnly_PreservesChainReceiver(t *testing.T) {
+	// Mirrors the Jungle4 production reproduction:
+	// atomicassets::createschema with shipload.gm as authorizer.
+	// Subscriber filtered on receivers=[shipload.gm] should still receive
+	// the action, but with wire Receiver=atomicassets (chain truth).
+	shipload := chain.StringToName("shipload.gm")
+	atomic := chain.StringToName("atomicassets")
+
+	createSchema := chain.ActionTrace{
+		Receiver: "atomicassets",
+		Act: chain.Action{
+			Account: "atomicassets",
+			Name:    "createschema",
+			Authorization: []chain.PermissionLevel{
+				{Actor: "shipload.gm", Permission: "active"},
+			},
+		},
+	}
+
+	catchup := &StreamCatchup{
+		reader: &stubBaseReader{traces: []chain.ActionTrace{createSchema}},
+		filter: ActionFilter{
+			Receivers: map[uint64]struct{}{shipload: {}},
+		},
+	}
+
+	var delivered []StreamedAction
+	sent, err := catchup.processBatch([]uint64{343972368}, shipload, nil, func(a StreamedAction) error {
+		delivered = append(delivered, a)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("processBatch error: %v", err)
+	}
+	if sent != 1 || len(delivered) != 1 {
+		t.Fatalf("expected 1 delivery, got sent=%d delivered=%d", sent, len(delivered))
+	}
+	got := delivered[0]
+	if got.Receiver != atomic {
+		t.Errorf("Receiver = %s, want atomicassets (chain receiver, not indexed-via shipload.gm)",
+			chain.NameToString(got.Receiver))
+	}
+	if got.Contract != atomic {
+		t.Errorf("Contract = %s, want atomicassets", chain.NameToString(got.Contract))
+	}
+	if got.Action != chain.StringToName("createschema") {
+		t.Errorf("Action = %s, want createschema", chain.NameToString(got.Action))
+	}
+}
diff --git a/services/actionindex/internal/stream_types.go b/services/actionindex/internal/stream_types.go
index 14b1ae2..2a4fc68 100644
--- a/services/actionindex/internal/stream_types.go
+++ b/services/actionindex/internal/stream_types.go
@@ -11,7 +11,7 @@ type ActionFilter struct {
 	Actions   map[uint64]struct{}
 }
 
-func (f *ActionFilter) Matches(action StreamedAction) bool {
+func (f *ActionFilter) Matches(action StreamedAction, matchedVia []uint64) bool {
 	if len(f.Contracts) == 0 && len(f.Receivers) == 0 {
 		return false
 	}
@@ -34,7 +34,8 @@ func (f *ActionFilter) Matches(action StreamedAction) bool {
 		return contractMatch
 	}
 
-	_, receiverMatch := f.Receivers[action.Receiver]
+	receiverMatch := anyInSet(matchedVia, f.Receivers)
+
 	if len(f.Contracts) == 0 {
 		return receiverMatch
 	}
@@ -42,6 +43,16 @@ func (f *ActionFilter) Matches(action StreamedAction) bool {
 	return contractMatch && receiverMatch
 }
 
+// anyInSet reports whether any element of keys is a key of m.
+func anyInSet(keys []uint64, m map[uint64]struct{}) bool {
+	for _, k := range keys {
+		if _, ok := m[k]; ok {
+			return true
+		}
+	}
+	return false
+}
+
 type StreamError struct {
 	Code    uint16
 	Message string
diff --git a/services/actionindex/internal/stream_types_test.go b/services/actionindex/internal/stream_types_test.go
index 8df33b5..3caea95 100644
--- a/services/actionindex/internal/stream_types_test.go
+++ b/services/actionindex/internal/stream_types_test.go
@@ -1,6 +1,10 @@
 package internal
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/greymass/roborovski/libraries/chain"
+)
 
 func TestActionFilter_Matches_ContractOnly(t *testing.T) {
 	contract := uint64(5000)
@@ -47,7 +51,7 @@ func TestActionFilter_Matches_ContractOnly(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := filter.Matches(tt.action)
+			got := filter.Matches(tt.action, []uint64{tt.action.Receiver})
 			if got != tt.want {
 				t.Errorf("Matches() = %v, want %v", got, tt.want)
 			}
@@ -101,7 +105,7 @@ func TestActionFilter_Matches_ContractAndReceiver(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := filter.Matches(tt.action)
+			got := filter.Matches(tt.action, []uint64{tt.action.Receiver})
 			if got != tt.want {
 				t.Errorf("Matches() = %v, want %v", got, tt.want)
 			}
@@ -144,7 +148,7 @@ func TestActionFilter_Matches_ReceiverOnly(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := filter.Matches(tt.action)
+			got := filter.Matches(tt.action, []uint64{tt.action.Receiver})
 			if got != tt.want {
 				t.Errorf("Matches() = %v, want %v", got, tt.want)
 			}
@@ -188,7 +192,7 @@ func TestActionFilter_Matches_WithActions(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := filter.Matches(tt.action)
+			got := filter.Matches(tt.action, []uint64{tt.action.Receiver})
 			if got != tt.want {
 				t.Errorf("Matches() = %v, want %v", got, tt.want)
 			}
@@ -205,7 +209,68 @@ func TestActionFilter_Matches_EmptyFilter(t *testing.T) {
 		Action:   6000,
 	}
 
-	if filter.Matches(action) {
+	if filter.Matches(action, []uint64{action.Receiver}) {
 		t.Error("Empty filter should not match any action")
 	}
 }
+
+func TestActionFilter_Matches_MatchedViaSingle(t *testing.T) {
+	a := chain.StringToName("alice")
+	f := ActionFilter{
+		Receivers: map[uint64]struct{}{a: {}},
+	}
+	act := StreamedAction{
+		Contract: chain.StringToName("eosio.token"),
+		Action:   chain.StringToName("transfer"),
+		Receiver: chain.StringToName("bob"), // chain truth != matched-via
+	}
+	if !f.Matches(act, []uint64{a}) {
+		t.Errorf("expected match: filter Receivers={alice}, matchedVia=[alice]")
+	}
+}
+
+func TestActionFilter_Matches_MatchedViaMulti(t *testing.T) {
+	a := chain.StringToName("alice")
+	b := chain.StringToName("bob")
+	f := ActionFilter{
+		Receivers: map[uint64]struct{}{a: {}},
+	}
+	act := StreamedAction{
+		Contract: chain.StringToName("eosio.token"),
+		Action:   chain.StringToName("transfer"),
+		Receiver: chain.StringToName("carol"),
+	}
+	if !f.Matches(act, []uint64{b, a}) {
+		t.Errorf("expected match: any element of matchedVia in Receivers should match")
+	}
+}
+
+func TestActionFilter_Matches_MatchedViaNone(t *testing.T) {
+	a := chain.StringToName("alice")
+	b := chain.StringToName("bob")
+	f := ActionFilter{
+		Receivers: map[uint64]struct{}{a: {}},
+	}
+	act := StreamedAction{
+		Contract: chain.StringToName("eosio.token"),
+		Action:   chain.StringToName("transfer"),
+		Receiver: a,
+	}
+	if f.Matches(act, []uint64{b}) {
+		t.Errorf("expected no match: matchedVia=[bob] is not in filter Receivers={alice}; chain Receiver is no longer consulted")
+	}
+}
+
+func TestActionFilter_Matches_ContractOnly_NilMatchedVia(t *testing.T) {
+	f := ActionFilter{
+		Contracts: map[uint64]struct{}{chain.StringToName("eosio.token"): {}},
+	}
+	act := StreamedAction{
+		Contract: chain.StringToName("eosio.token"),
+		Action:   chain.StringToName("transfer"),
+		Receiver: chain.StringToName("alice"),
+	}
+	if !f.Matches(act, nil) {
+		t.Errorf("expected match on contract-only filter regardless of matchedVia")
+	}
+}
diff --git a/services/actionindex/internal/sync.go b/services/actionindex/internal/sync.go
index 29ed015..d2743c4 100644
--- a/services/actionindex/internal/sync.go
+++ b/services/actionindex/internal/sync.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"encoding/hex"
 	"fmt"
+	"slices"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -299,16 +300,35 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 		totalStart = time.Now()
 	}
 
+	type seqGroup struct {
+		firstIndex int
+		matchedVia []uint64
+	}
+	groups := make(map[uint64]*seqGroup, len(block.Actions))
+	seqOrder := make([]uint64, 0, len(block.Actions))
+	for i := range block.Actions {
+		a := &block.Actions[i]
+		g, exists := groups[a.GlobalSeq]
+		if !exists {
+			g = &seqGroup{firstIndex: i}
+			groups[a.GlobalSeq] = g
+			seqOrder = append(seqOrder, a.GlobalSeq)
+		}
+		g.matchedVia = append(g.matchedVia, a.Account)
+	}
+	slices.Sort(seqOrder)
+
 	var filterStart time.Time
 	if p.syncer.config.QueryTrace {
 		filterStart = time.Now()
 	}
 
-	var matchingActions []int
-	for i := range block.Actions {
-		a := &block.Actions[i]
-		if p.syncer.broadcaster.CouldMatch(a.Contract, a.Action, a.Account) {
-			matchingActions = append(matchingActions, i)
+	matchingSeqs := make([]uint64, 0, len(seqOrder))
+	for _, seq := range seqOrder {
+		g := groups[seq]
+		a := &block.Actions[g.firstIndex]
+		if p.syncer.broadcaster.CouldMatch(a.Contract, a.Action, g.matchedVia) {
+			matchingSeqs = append(matchingSeqs, seq)
 		}
 	}
 
@@ -317,17 +337,27 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 		filterDur = time.Since(filterStart)
 	}
 
-	if len(matchingActions) == 0 {
+	if len(matchingSeqs) == 0 {
 		if p.syncer.config.QueryTrace {
-			logger.Printf("debug-stream", "block=%d actions=%d matched=0 subs=%d filter=%v",
-				block.BlockNum, len(block.Actions), subCount, filterDur)
+			logger.Printf("debug-stream", "block=%d actions=%d unique_seqs=%d matched=0 subs=%d filter=%v",
+				block.BlockNum, len(block.Actions), len(seqOrder), subCount, filterDur)
 		}
 		return nil
 	}
 
-	var dataStart time.Time
-	if p.syncer.config.QueryTrace {
-		dataStart = time.Now()
+	sendOne := func(seq uint64, g *seqGroup, data []byte, cpuUs, netWords uint32) bool {
+		a := &block.Actions[g.firstIndex]
+		return p.syncer.broadcaster.Broadcast(StreamedAction{
+			GlobalSeq:     seq,
+			BlockNum:      block.BlockNum,
+			BlockTime:     block.BlockTime,
+			Contract:      a.Contract,
+			Action:        a.Action,
+			Receiver:      a.Receiver,
+			ActionData:    data,
+			CpuUsageUs:    cpuUs,
+			NetUsageWords: netWords,
+		}, g.matchedVia)
 	}
 
 	var delivered int
@@ -335,47 +365,26 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 	source := "inline"
 
 	if block.HasActionData() {
-		var dataDur time.Duration
-		if p.syncer.config.QueryTrace {
-			dataDur = time.Since(dataStart)
-		}
-
 		var sendStart time.Time
 		if p.syncer.config.QueryTrace {
 			sendStart = time.Now()
 		}
 
-		for _, idx := range matchingActions {
-			a := &block.Actions[idx]
-			actionData := block.GetActionDataBySeq(a.GlobalSeq)
-			cpuUs, netWords := block.GetResourceUsage(a.GlobalSeq)
-			if p.syncer.broadcaster.Broadcast(StreamedAction{
-				GlobalSeq:     a.GlobalSeq,
-				BlockNum:      block.BlockNum,
-				BlockTime:     block.BlockTime,
-				Contract:      a.Contract,
-				Action:        a.Action,
-				Receiver:      a.Account,
-				ActionData:    actionData,
-				CpuUsageUs:    cpuUs,
-				NetUsageWords: netWords,
-			}) {
+		for _, seq := range matchingSeqs {
+			g := groups[seq]
+			data := block.GetActionDataBySeq(seq)
+			cpuUs, netWords := block.GetResourceUsage(seq)
+			if sendOne(seq, g, data, cpuUs, netWords) {
 				delivered++
 			}
 		}
 
 		if p.syncer.config.QueryTrace {
 			sendDur = time.Since(sendStart)
-			_ = dataDur
 		}
 	} else {
 		source = "fetched"
-		seqs := make([]uint64, len(matchingActions))
-		for i, idx := range matchingActions {
-			seqs[i] = block.Actions[idx].GlobalSeq
-		}
-
-		traces, _, err := p.fetchActionDataWithRetry(block.BlockNum, seqs)
+		traces, _, err := p.fetchActionDataWithRetry(block.BlockNum, matchingSeqs)
 		if err != nil {
 			p.syncer.broadcaster.BroadcastError(ActionErrorDataInconsistent, err.Error())
 			return err
@@ -387,22 +396,12 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 		}
 
 		for i, at := range traces {
-			idx := matchingActions[i]
-			var actionData []byte
+			seq := matchingSeqs[i]
+			var data []byte
 			if at.Act.Data != "" {
-				actionData, _ = hex.DecodeString(at.Act.Data)
+				data, _ = hex.DecodeString(at.Act.Data)
 			}
-			if p.syncer.broadcaster.Broadcast(StreamedAction{
-				GlobalSeq:     seqs[i],
-				BlockNum:      block.BlockNum,
-				BlockTime:     block.BlockTime,
-				Contract:      block.Actions[idx].Contract,
-				Action:        block.Actions[idx].Action,
-				Receiver:      block.Actions[idx].Account,
-				ActionData:    actionData,
-				CpuUsageUs:    at.CpuUsageUs,
-				NetUsageWords: at.NetUsageWords,
-			}) {
+			if sendOne(seq, groups[seq], data, at.CpuUsageUs, at.NetUsageWords) {
 				delivered++
 			}
 		}
@@ -414,8 +413,8 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error
 
 	if p.syncer.config.QueryTrace {
 		totalDur := time.Since(totalStart)
-		logger.Printf("debug-stream", "block=%d actions=%d matched=%d delivered=%d subs=%d (%s) total=%v filter=%v send=%v",
-			block.BlockNum, len(block.Actions), len(matchingActions), delivered, subCount, source, totalDur, filterDur, sendDur)
+		logger.Printf("debug-stream", "block=%d actions=%d unique_seqs=%d matched=%d delivered=%d subs=%d (%s) total=%v filter=%v send=%v",
+			block.BlockNum, len(block.Actions), len(seqOrder), len(matchingSeqs), delivered, subCount, source, totalDur, filterDur, sendDur)
 	}
 
 	return nil
diff --git a/services/actionindex/internal/sync_broadcast_test.go b/services/actionindex/internal/sync_broadcast_test.go
new file mode 100644
index 0000000..5b8948b
--- /dev/null
+++ b/services/actionindex/internal/sync_broadcast_test.go
@@ -0,0 +1,304 @@
+package internal
+
+import (
+	"testing"
+	"time"
+
+	"github.com/greymass/roborovski/libraries/chain"
+	"github.com/greymass/roborovski/libraries/corereader"
+)
+
+
+// stubReaderForBroadcast satisfies corereader.Reader minimally for broadcastActions tests.
+// Only GetActionsByGlobalSeqs is called from the fetched-traces sub-path.
+type stubReaderForBroadcast struct {
+	corereader.Reader
+	bySeq map[uint64]chain.ActionTrace
+}
+
+func (s *stubReaderForBroadcast) GetActionsByGlobalSeqs(seqs []uint64) ([]chain.ActionTrace, *corereader.FetchTimings, error) {
+	out := make([]chain.ActionTrace, 0, len(seqs))
+	for _, seq := range seqs {
+		out = append(out, s.bySeq[seq])
+	}
+	return out, &corereader.FetchTimings{}, nil
+}
+
+// drainExactly drains exactly n actions from the subscription. Returns nil if
+// n actions don't arrive within a short timeout. Use this for the happy path
+// where you know the expected count.
+func drainExactly(t *testing.T, sub *Subscription, n int) []StreamedAction {
+	t.Helper()
+	got := make([]StreamedAction, 0, n)
+	deadline := time.After(200 * time.Millisecond)
+	for len(got) < n {
+		select {
+		case a, ok := <-sub.sendCh:
+			if !ok {
+				return got
+			}
+			got = append(got, a)
+		case <-deadline:
+			return got
+		}
+	}
+	return got
+}
+
+// assertNoMore verifies that no additional actions arrive in a short window.
+// Use after drainExactly when you need to prove the count is tight (not just >=).
+func assertNoMore(t *testing.T, sub *Subscription, window time.Duration) {
+	t.Helper()
+	select {
+	case extra := <-sub.sendCh:
+		t.Fatalf("unexpected extra delivery: %+v", extra)
+	case <-time.After(window):
+	}
+}
+
+func TestBroadcastActions_AuthorizerOnly_UsesChainReceiver(t *testing.T) {
+	atomic := chain.StringToName("atomicassets")
+	createschema := chain.StringToName("createschema")
+	shipload := chain.StringToName("shipload.gm")
+	const seq = uint64(343972368)
+
+	broadcaster := NewActionBroadcaster()
+	broadcaster.SetLiveMode(true)
+	sub := broadcaster.Subscribe(ActionFilter{
+		Receivers: map[uint64]struct{}{shipload: {}},
+	})
+
+	reader := &stubReaderForBroadcast{
+		bySeq: map[uint64]chain.ActionTrace{
+			seq: {
+				BlockNum: 1, BlockTime: "1970-01-01T00:16:40.000",
+				Receiver: "atomicassets",
+				Act: chain.Action{
+					Account: "atomicassets", Name: "createschema",
+					Authorization: []chain.PermissionLevel{
+						{Actor: "shipload.gm", Permission: "active"},
+					},
+				},
+			},
+		},
+	}
+
+	block := corereader.Block{
+		BlockNum:  1,
+		BlockTime: 1000,
+		MaxSeq:    seq,
+		Actions: []corereader.Action{
+			// Authorizer-only entry: shipload.gm indexed because it authorized
+			{Account: shipload, Contract: atomic, Action: createschema, GlobalSeq: seq, Receiver: atomic, IsAuthorizer: true},
+		},
+	}
+
+	syncer := &Syncer{broadcaster: broadcaster, reader: reader, config: &Config{}}
+	proc := NewAccountHistoryProcessor(syncer)
+	if err := proc.broadcastActions(block); err != nil {
+		t.Fatalf("broadcastActions error: %v", err)
+	}
+
+	delivered := drainExactly(t, sub, 1)
+	if len(delivered) != 1 {
+		t.Fatalf("expected 1 delivery, got %d", len(delivered))
+	}
+	if delivered[0].Receiver != atomic {
+		t.Errorf("Receiver = %s, want atomicassets (chain receiver)",
+			chain.NameToString(delivered[0].Receiver))
+	}
+	if delivered[0].GlobalSeq != seq {
+		t.Errorf("GlobalSeq = %d, want %d", delivered[0].GlobalSeq, seq)
+	}
+	if delivered[0].Contract != atomic {
+		t.Errorf("Contract = %s, want atomicassets", chain.NameToString(delivered[0].Contract))
+	}
+	if delivered[0].Action != createschema {
+		t.Errorf("Action = %s, want createschema", chain.NameToString(delivered[0].Action))
+	}
+}
+
+func TestBroadcastActions_DedupesByGlobalSeq(t *testing.T) {
+	contract := chain.StringToName("eosio.token")
+	transfer := chain.StringToName("transfer")
+	alice := chain.StringToName("alice")
+	bob := chain.StringToName("bob")
+	const seq = uint64(5000)
+
+	broadcaster := NewActionBroadcaster()
+	broadcaster.SetLiveMode(true)
+	// Sub matches both alice (chain receiver) AND bob (authorizer) on the same seq.
+	sub := broadcaster.Subscribe(ActionFilter{
+		Receivers: map[uint64]struct{}{alice: {}, bob: {}},
+	})
+
+	reader := &stubReaderForBroadcast{
+		bySeq: map[uint64]chain.ActionTrace{
+			seq: {
+				BlockNum: 1, BlockTime: "1970-01-01T00:16:40.000",
+				Receiver: "alice",
+				Act: chain.Action{
+					Account: "eosio.token", Name: "transfer",
+					Authorization: []chain.PermissionLevel{
+						{Actor: "bob", Permission: "active"},
+					},
+				},
+			},
+		},
+	}
+
+	block := corereader.Block{
+		BlockNum:  1,
+		BlockTime: 1000,
+		MaxSeq:    seq,
+		Actions: []corereader.Action{
+			// Two index entries for the same seq: receiver-path + authorizer-path
+			{Account: alice, Contract: contract, Action: transfer, GlobalSeq: seq, Receiver: alice, IsAuthorizer: false},
+			{Account: bob, Contract: contract, Action: transfer, GlobalSeq: seq, Receiver: alice, IsAuthorizer: true},
+		},
+	}
+
+	syncer := &Syncer{broadcaster: broadcaster, reader: reader, config: &Config{}}
+	proc := NewAccountHistoryProcessor(syncer)
+	if err := proc.broadcastActions(block); err != nil {
+		t.Fatalf("broadcastActions error: %v", err)
+	}
+
+	delivered := drainExactly(t, sub, 1)
+	assertNoMore(t, sub, 50*time.Millisecond)
+	if len(delivered) != 1 {
+		t.Fatalf("expected exactly 1 delivery (dedup), got %d", len(delivered))
+	}
+	if delivered[0].Receiver != alice {
+		t.Errorf("Receiver = %s, want alice (chain receiver)", chain.NameToString(delivered[0].Receiver))
+	}
+	if delivered[0].GlobalSeq != seq {
+		t.Errorf("GlobalSeq = %d, want %d", delivered[0].GlobalSeq, seq)
+	}
+	if delivered[0].Contract != contract {
+		t.Errorf("Contract = %s, want eosio.token", chain.NameToString(delivered[0].Contract))
+	}
+	if delivered[0].Action != transfer {
+		t.Errorf("Action = %s, want transfer", chain.NameToString(delivered[0].Action))
+	}
+}
+
+func TestBroadcastActions_InlinePath_UsesChainReceiver(t *testing.T) {
+	// Exercises the inline-data branch (block.HasActionData() == true), which
+	// is the production hot path for live sync. Mirrors AuthorizerOnly_UsesChainReceiver
+	// but constructs a RawBlock with action data so the broadcaster uses the
+	// inline branch instead of the fetched-traces fallback.
+	atomic := chain.StringToName("atomicassets")
+	createschema := chain.StringToName("createschema")
+	shipload := chain.StringToName("shipload.gm")
+	const seq = uint64(343972368)
+
+	broadcaster := NewActionBroadcaster()
+	broadcaster.SetLiveMode(true)
+	sub := broadcaster.Subscribe(ActionFilter{
+		Receivers: map[uint64]struct{}{shipload: {}},
+	})
+
+	rawData := []byte{0xDE, 0xAD, 0xBE, 0xEF}
+	raw := &corereader.RawBlock{
+		BlockNum:  1,
+		BlockTime: 1000,
+		Actions: []corereader.CanonicalAction{
+			{GlobalSeqUint64: seq, DataIndex: 0, CpuUsageUs: 42, NetUsageWords: 7},
+		},
+	}
+	raw.SetActionData(rawData, []uint32{0}, []uint32{uint32(len(rawData))})
+
+	block := corereader.Block{
+		BlockNum:  1,
+		BlockTime: 1000,
+		MaxSeq:    seq,
+		Actions: []corereader.Action{
+			{Account: shipload, Contract: atomic, Action: createschema, GlobalSeq: seq, Receiver: atomic, IsAuthorizer: true},
+		},
+	}
+	block.SetRawBlock(raw)
+
+	// Reader stub isn't used in the inline path but Syncer requires a non-nil one.
+	syncer := &Syncer{broadcaster: broadcaster, reader: &stubReaderForBroadcast{}, config: &Config{}}
+	proc := NewAccountHistoryProcessor(syncer)
+	if err := proc.broadcastActions(block); err != nil {
+		t.Fatalf("broadcastActions error: %v", err)
+	}
+
+	delivered := drainExactly(t, sub, 1)
+	if len(delivered) != 1 {
+		t.Fatalf("expected 1 delivery, got %d", len(delivered))
+	}
+	got := delivered[0]
+	if got.Receiver != atomic {
+		t.Errorf("Receiver = %s, want atomicassets (chain receiver)", chain.NameToString(got.Receiver))
+	}
+	if got.GlobalSeq != seq {
+		t.Errorf("GlobalSeq = %d, want %d", got.GlobalSeq, seq)
+	}
+	if got.Contract != atomic {
+		t.Errorf("Contract = %s, want atomicassets", chain.NameToString(got.Contract))
+	}
+	if got.Action != createschema {
+		t.Errorf("Action = %s, want createschema", chain.NameToString(got.Action))
+	}
+	if got.CpuUsageUs != 42 || got.NetUsageWords != 7 {
+		t.Errorf("resource usage = cpu=%d net=%d, want cpu=42 net=7", got.CpuUsageUs, got.NetUsageWords)
+	}
+}
+
+func TestBroadcastActions_ContractOnlyFilter_DeliversAuthorizerIndexed(t *testing.T) {
+	atomic := chain.StringToName("atomicassets")
+	createschema := chain.StringToName("createschema")
+	shipload := chain.StringToName("shipload.gm")
+	const seq = uint64(343972368)
+
+	broadcaster := NewActionBroadcaster()
+	broadcaster.SetLiveMode(true)
+	sub := broadcaster.Subscribe(ActionFilter{
+		Contracts: map[uint64]struct{}{atomic: {}},
+	})
+
+	reader := &stubReaderForBroadcast{
+		bySeq: map[uint64]chain.ActionTrace{
+			seq: {
+				BlockNum: 1, BlockTime: "1970-01-01T00:16:40.000",
+				Receiver: "atomicassets",
+				Act: chain.Action{
+					Account: "atomicassets", Name: "createschema",
+					Authorization: []chain.PermissionLevel{
+						{Actor: "shipload.gm", Permission: "active"},
+					},
+				},
+			},
+		},
+	}
+
+	block := corereader.Block{
+		BlockNum:  1,
+		BlockTime: 1000,
+		MaxSeq:    seq,
+		Actions: []corereader.Action{
+			{Account: shipload, Contract: atomic, Action: createschema, GlobalSeq: seq, Receiver: atomic, IsAuthorizer: true},
+		},
+	}
+
+	syncer := &Syncer{broadcaster: broadcaster, reader: reader, config: &Config{}}
+	proc := NewAccountHistoryProcessor(syncer)
+	if err := proc.broadcastActions(block); err != nil {
+		t.Fatalf("broadcastActions error: %v", err)
+	}
+
+	delivered := drainExactly(t, sub, 1)
+	if len(delivered) != 1 {
+		t.Fatalf("expected 1 delivery to contract-only subscriber, got %d", len(delivered))
+	}
+	got := delivered[0]
+	if got.Contract != atomic {
+		t.Errorf("Contract = %s, want atomicassets", chain.NameToString(got.Contract))
+	}
+	if got.Receiver != atomic {
+		t.Errorf("Receiver = %s, want atomicassets", chain.NameToString(got.Receiver))
+	}
+}

From 6bc2d4663af3886bdd6adebfc62609388a6dbf38 Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Fri, 22 May 2026 19:31:46 -0700
Subject: [PATCH 10/11] Select proper block from speculative entires list

---
 libraries/tracereader/tracefiles.go           |  82 +++++-------
 libraries/tracereader/tracefiles_fork_test.go | 123 ++++++++++++++++++
 2 files changed, 159 insertions(+), 46 deletions(-)
 create mode 100644 libraries/tracereader/tracefiles_fork_test.go

diff --git a/libraries/tracereader/tracefiles.go b/libraries/tracereader/tracefiles.go
index a462481..28d158e 100644
--- a/libraries/tracereader/tracefiles.go
+++ b/libraries/tracereader/tracefiles.go
@@ -148,8 +148,12 @@ func GetRawBlocksWithMetadata(blockNum uint32, limit int, conf *Config) ([]RawBl
 	}
 	defer sliceIndex.Close()
 
-	var found bool = false
-	var offsets []uint64 = make([]uint64, 0)
+	type blockIdxEntry struct {
+		number uint32
+		offset uint64
+	}
+	var blockEntries []blockIdxEntry
+	latestIdx := make(map[uint32]int)
 
 	var entry IndexEntryVariant
 	dec := chain.NewDecoder(sliceIndex)
@@ -161,34 +165,19 @@ func GetRawBlocksWithMetadata(blockNum uint32, limit int, conf *Config) ([]RawBl
 			}
 			return nil, err
 		}
-
-		if entry.Block != nil && entry.Block.Number >= blockNum {
-			if entry.Block.Number == blockNum {
-				found = true
-			}
-			offsets = append(offsets, entry.Block.Offset)
-			if limit > 0 && len(offsets) >= limit {
-				err = dec.DecodeVariant(&entry)
-				if err != io.EOF && entry.Block != nil {
-					offsets = append(offsets, entry.Block.Offset)
-				}
-				break
-			}
+		if entry.Block != nil {
+			blockEntries = append(blockEntries, blockIdxEntry{
+				number: entry.Block.Number,
+				offset: entry.Block.Offset,
+			})
+			latestIdx[entry.Block.Number] = len(blockEntries) - 1
 		}
 	}
 
-	if !found {
+	if _, ok := latestIdx[blockNum]; !ok {
 		return nil, ErrNotFound
 	}
 
-	numBlocks := limit
-	if limit <= 0 || len(offsets) < limit {
-		numBlocks = len(offsets)
-		if numBlocks > 0 {
-			numBlocks--
-		}
-	}
-
 	slice, err := os.OpenFile(fmt.Sprintf("%s/trace_%s.log", conf.Dir, stride), os.O_RDONLY, 0)
 	if err != nil {
 		return nil, err
@@ -201,49 +190,50 @@ func GetRawBlocksWithMetadata(blockNum uint32, limit int, conf *Config) ([]RawBl
 	}
 	traceFileSize := sliceInfo.Size()
 
-	rawBlocks := make([]RawBlockData, numBlocks)
-
-	for i := 0; i < numBlocks; i++ {
+	var rawBlocks []RawBlockData
+	for k := uint32(0); limit <= 0 || k < uint32(limit); k++ {
+		bn := blockNum + k
+		idx, ok := latestIdx[bn]
+		if !ok {
+			break
+		}
+		offset := blockEntries[idx].offset
 		var blockSize uint64
-		if i+1 < len(offsets) {
-			blockSize = offsets[i+1] - offsets[i]
+		if idx+1 < len(blockEntries) {
+			blockSize = blockEntries[idx+1].offset - offset
 		} else {
-			blockSize = uint64(traceFileSize) - offsets[i]
+			blockSize = uint64(traceFileSize) - offset
 		}
 
-		currentBlockNum := blockNum + uint32(i)
-
-		if int64(offsets[i]) >= traceFileSize {
+		if int64(offset) >= traceFileSize {
 			return nil, fmt.Errorf("block %d: offset %d exceeds trace file size %d (stride=%s)",
-				currentBlockNum, offsets[i], traceFileSize, stride)
+				bn, offset, traceFileSize, stride)
 		}
 
-		if int64(offsets[i]+blockSize) > traceFileSize {
-			actualAvailable := traceFileSize - int64(offsets[i])
+		if int64(offset+blockSize) > traceFileSize {
+			actualAvailable := traceFileSize - int64(offset)
 			return nil, fmt.Errorf("block %d: expected %d bytes at offset %d but only %d bytes available in trace file (size=%d, stride=%s)",
-				currentBlockNum, blockSize, offsets[i], actualAvailable, traceFileSize, stride)
+				bn, blockSize, offset, actualAvailable, traceFileSize, stride)
 		}
 
-		_, err = slice.Seek(int64(offsets[i]), io.SeekStart)
-		if err != nil {
-			return nil, fmt.Errorf("failed to seek to block %d offset %d: %w", currentBlockNum, offsets[i], err)
+		if _, err := slice.Seek(int64(offset), io.SeekStart); err != nil {
+			return nil, fmt.Errorf("failed to seek to block %d offset %d: %w", bn, offset, err)
 		}
 
 		rawBytes := make([]byte, blockSize)
 		n, err := slice.Read(rawBytes)
 		if err != nil && err != io.EOF {
-			return nil, fmt.Errorf("failed to read raw bytes for block %d: %w", currentBlockNum, err)
+			return nil, fmt.Errorf("failed to read raw bytes for block %d: %w", bn, err)
 		}
-
 		if uint64(n) != blockSize {
 			return nil, fmt.Errorf("block %d: read %d bytes but expected %d (offset=%d, fileSize=%d, stride=%s)",
-				currentBlockNum, n, blockSize, offsets[i], traceFileSize, stride)
+				bn, n, blockSize, offset, traceFileSize, stride)
 		}
 
-		rawBlocks[i] = RawBlockData{
-			BlockNum: currentBlockNum,
+		rawBlocks = append(rawBlocks, RawBlockData{
+			BlockNum: bn,
 			RawBytes: rawBytes,
-		}
+		})
 	}
 
 	return rawBlocks, nil
diff --git a/libraries/tracereader/tracefiles_fork_test.go b/libraries/tracereader/tracefiles_fork_test.go
new file mode 100644
index 0000000..1392926
--- /dev/null
+++ b/libraries/tracereader/tracefiles_fork_test.go
@@ -0,0 +1,123 @@
+package tracereader
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+type forkScenarioEntry struct {
+	Number uint32
+	Marker byte
+	Size   int
+}
+
+func writeForkScenario(t *testing.T, dir string, lo, hi uint32, entries []forkScenarioEntry, libBlock uint32) (string, string) {
+	t.Helper()
+	stride := fmt.Sprintf("%010d-%010d", lo, hi)
+	indexPath := filepath.Join(dir, "trace_index_"+stride+".log")
+	dataPath := filepath.Join(dir, "trace_"+stride+".log")
+
+	indexBuf := new(bytes.Buffer)
+	dataBuf := new(bytes.Buffer)
+	binary.Write(indexBuf, binary.LittleEndian, HeaderVersion)
+
+	offset := uint64(0)
+	for i, e := range entries {
+		indexBuf.WriteByte(0) // variant tag = BlockEntryV0
+		var id [32]byte
+		id[0] = e.Marker
+		id[1] = byte(i)
+		indexBuf.Write(id[:])
+		binary.Write(indexBuf, binary.LittleEndian, e.Number)
+		binary.Write(indexBuf, binary.LittleEndian, offset)
+
+		for j := 0; j < e.Size; j++ {
+			dataBuf.WriteByte(e.Marker)
+		}
+		offset += uint64(e.Size)
+	}
+
+	indexBuf.WriteByte(1) // variant tag = LibEntryV0
+	binary.Write(indexBuf, binary.LittleEndian, libBlock)
+
+	if err := os.WriteFile(indexPath, indexBuf.Bytes(), 0644); err != nil {
+		t.Fatalf("write index: %v", err)
+	}
+	if err := os.WriteFile(dataPath, dataBuf.Bytes(), 0644); err != nil {
+		t.Fatalf("write data: %v", err)
+	}
+	return indexPath, dataPath
+}
+
+func TestGetRawBlocksWithMetadata_PicksCanonicalAcrossForks(t *testing.T) {
+	dir := t.TempDir()
+	conf := &Config{Stride: 500, Dir: dir}
+
+	writeForkScenario(t, dir, 0, 500, []forkScenarioEntry{
+		{Number: 100, Marker: 'A', Size: 64},
+		{Number: 101, Marker: 'F', Size: 64}, // forked 101
+		{Number: 101, Marker: 'C', Size: 64}, // canonical 101 (last entry wins)
+		{Number: 102, Marker: 'D', Size: 64},
+	}, 110)
+
+	rawBlocks, err := GetRawBlocksWithMetadata(101, 1, conf)
+	if err != nil {
+		t.Fatalf("GetRawBlocksWithMetadata error: %v", err)
+	}
+	if len(rawBlocks) != 1 {
+		t.Fatalf("expected 1 raw block, got %d", len(rawBlocks))
+	}
+	got := rawBlocks[0]
+	if got.BlockNum != 101 {
+		t.Fatalf("BlockNum = %d, want 101", got.BlockNum)
+	}
+	if len(got.RawBytes) != 64 {
+		t.Fatalf("RawBytes len = %d, want 64", len(got.RawBytes))
+	}
+	for i, b := range got.RawBytes {
+		if b != 'C' {
+			t.Fatalf("RawBytes[%d] = %q, want 'C' (canonical). Forked marker 'F' indicates the reader picked the first-written (orphaned) trace_index entry instead of the last (canonical) one.", i, b)
+		}
+	}
+}
+
+func TestGetRawBlocksWithMetadata_PicksCanonicalForRangeAcrossForks(t *testing.T) {
+	dir := t.TempDir()
+	conf := &Config{Stride: 500, Dir: dir}
+
+	writeForkScenario(t, dir, 0, 500, []forkScenarioEntry{
+		{Number: 100, Marker: 'A', Size: 32},
+		{Number: 101, Marker: 'F', Size: 32}, // forked 101
+		{Number: 101, Marker: 'C', Size: 32}, // canonical 101
+		{Number: 102, Marker: 'D', Size: 32},
+	}, 110)
+
+	rawBlocks, err := GetRawBlocksWithMetadata(100, 3, conf)
+	if err != nil {
+		t.Fatalf("GetRawBlocksWithMetadata error: %v", err)
+	}
+	if len(rawBlocks) != 3 {
+		t.Fatalf("expected 3 raw blocks (100, 101, 102), got %d", len(rawBlocks))
+	}
+
+	want := []struct {
+		num    uint32
+		marker byte
+	}{
+		{100, 'A'},
+		{101, 'C'},
+		{102, 'D'},
+	}
+	for i, w := range want {
+		if rawBlocks[i].BlockNum != w.num {
+			t.Errorf("rawBlocks[%d].BlockNum = %d, want %d", i, rawBlocks[i].BlockNum, w.num)
+		}
+		if len(rawBlocks[i].RawBytes) == 0 || rawBlocks[i].RawBytes[0] != w.marker {
+			t.Errorf("rawBlocks[%d] marker = %q, want %q", i, rawBlocks[i].RawBytes[0], w.marker)
+		}
+	}
+}

From 8ed6ab57a2b1cdaa620c7b4e345b8736b7a4721d Mon Sep 17 00:00:00 2001
From: aaroncox <aaron.cox@greymass.com>
Date: Fri, 22 May 2026 20:58:41 -0700
Subject: [PATCH 11/11] Restoring CGO_ENABLED and defensively cache when
 disabled

---
 Makefile                            | 40 ++++++++++++++---------------
 libraries/compression/zstd_nocgo.go | 32 +++++++++++++++++++++--
 2 files changed, 50 insertions(+), 22 deletions(-)

diff --git a/Makefile b/Makefile
index 34e8349..afe7a93 100644
--- a/Makefile
+++ b/Makefile
@@ -136,8 +136,8 @@ endef
 .PHONY: build/actionindex
 build/actionindex:
 	@if [ "$(call needs_rebuild,actionindex,services/actionindex)" = "1" ]; then \
-		echo "==> Building actionindex (CGO_ENABLED=0)"; \
-		CGO_ENABLED=0 go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/actionindex ./services/actionindex/cmd/actionindex; \
+		echo "==> Building actionindex"; \
+		go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/actionindex ./services/actionindex/cmd/actionindex; \
 	else \
 		echo "==> actionindex is up to date"; \
 	fi
@@ -145,8 +145,8 @@ build/actionindex:
 .PHONY: build/coreverify
 build/coreverify:
 	@if [ "$(call needs_rebuild,coreverify,services/coreverify)" = "1" ]; then \
-		echo "==> Building coreverify (CGO_ENABLED=0)"; \
-		CGO_ENABLED=0 go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/coreverify ./services/coreverify/cmd/coreverify; \
+		echo "==> Building coreverify"; \
+		go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/coreverify ./services/coreverify/cmd/coreverify; \
 	else \
 		echo "==> coreverify is up to date"; \
 	fi
@@ -154,8 +154,8 @@ build/coreverify:
 .PHONY: build/apiproxy
 build/apiproxy:
 	@if [ "$(call needs_rebuild,apiproxy,services/apiproxy)" = "1" ]; then \
-		echo "==> Building apiproxy (CGO_ENABLED=0)"; \
-		CGO_ENABLED=0 go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/apiproxy ./services/apiproxy/cmd/apiproxy; \
+		echo "==> Building apiproxy"; \
+		go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/apiproxy ./services/apiproxy/cmd/apiproxy; \
 	else \
 		echo "==> apiproxy is up to date"; \
 	fi
@@ -163,8 +163,8 @@ build/apiproxy:
 .PHONY: build/coreindex
 build/coreindex:
 	@if [ "$(call needs_rebuild,coreindex,services/coreindex)" = "1" ]; then \
-		echo "==> Building coreindex (CGO_ENABLED=0)"; \
-		CGO_ENABLED=0 go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/coreindex ./services/coreindex/cmd/coreindex; \
+		echo "==> Building coreindex"; \
+		go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/coreindex ./services/coreindex/cmd/coreindex; \
 	else \
 		echo "==> coreindex is up to date"; \
 	fi
@@ -172,8 +172,8 @@ build/coreindex:
 .PHONY: build/txindex
 build/txindex:
 	@if [ "$(call needs_rebuild,txindex,services/txindex)" = "1" ]; then \
-		echo "==> Building txindex (CGO_ENABLED=0)"; \
-		CGO_ENABLED=0 go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/txindex ./services/txindex/cmd/txindex; \
+		echo "==> Building txindex"; \
+		go build -ldflags "-X main.Version=$(VERSION)" -o $(BINDIR)/txindex ./services/txindex/cmd/txindex; \
 	else \
 		echo "==> txindex is up to date"; \
 	fi
@@ -184,28 +184,28 @@ build/txindex:
 
 .PHONY: install/actionindex
 install/actionindex:
-	@echo "==> Installing actionindex (CGO_ENABLED=0)"
-	@CGO_ENABLED=0 go install ./services/actionindex/cmd/actionindex
+	@echo "==> Installing actionindex"
+	@go install ./services/actionindex/cmd/actionindex
 
 .PHONY: install/coreverify
 install/coreverify:
-	@echo "==> Installing coreverify (CGO_ENABLED=0)"
-	@CGO_ENABLED=0 go install ./services/coreverify/cmd/coreverify
+	@echo "==> Installing coreverify"
+	@go install ./services/coreverify/cmd/coreverify
 
 .PHONY: install/apiproxy
 install/apiproxy:
-	@echo "==> Installing apiproxy (CGO_ENABLED=0)"
-	@CGO_ENABLED=0 go install ./services/apiproxy/cmd/apiproxy
+	@echo "==> Installing apiproxy"
+	@go install ./services/apiproxy/cmd/apiproxy
 
 .PHONY: install/coreindex
 install/coreindex:
-	@echo "==> Installing coreindex (CGO_ENABLED=0)"
-	@CGO_ENABLED=0 go install ./services/coreindex/cmd/coreindex
+	@echo "==> Installing coreindex"
+	@go install ./services/coreindex/cmd/coreindex
 
 .PHONY: install/txindex
 install/txindex:
-	@echo "==> Installing txindex (CGO_ENABLED=0)"
-	@CGO_ENABLED=0 go install ./services/txindex/cmd/txindex
+	@echo "==> Installing txindex"
+	@go install ./services/txindex/cmd/txindex
 
 # =============================================================================
 # Release Targets
diff --git a/libraries/compression/zstd_nocgo.go b/libraries/compression/zstd_nocgo.go
index cd67292..2fd6fdd 100644
--- a/libraries/compression/zstd_nocgo.go
+++ b/libraries/compression/zstd_nocgo.go
@@ -30,12 +30,40 @@ func putDecoder(d *zstd.Decoder) {
 	decoderPool.Put(d)
 }
 
+var (
+	encoderMu    sync.RWMutex
+	encoderCache = map[int]*zstd.Encoder{}
+)
+
+func getEncoder(level int) (*zstd.Encoder, error) {
+	encoderMu.RLock()
+	enc, ok := encoderCache[level]
+	encoderMu.RUnlock()
+	if ok {
+		return enc, nil
+	}
+
+	encoderMu.Lock()
+	defer encoderMu.Unlock()
+	if enc, ok = encoderCache[level]; ok {
+		return enc, nil
+	}
+	enc, err := zstd.NewWriter(nil,
+		zstd.WithEncoderLevel(zstd.EncoderLevelFromZstd(level)),
+		zstd.WithEncoderConcurrency(1),
+	)
+	if err != nil {
+		return nil, err
+	}
+	encoderCache[level] = enc
+	return enc, nil
+}
+
 func ZstdCompressLevel(dst, src []byte, level int) ([]byte, error) {
-	enc, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.EncoderLevelFromZstd(level)))
+	enc, err := getEncoder(level)
 	if err != nil {
 		return nil, err
 	}
-	defer enc.Close()
 	return enc.EncodeAll(src, dst[:0]), nil
 }