From 42fd5e1c523dd4a6b19f0af716ba613f88717224 Mon Sep 17 00:00:00 2001 From: aaroncox Date: Sat, 6 Jun 2026 14:03:58 -0700 Subject: [PATCH 1/2] handle trace live edge reads and enforce gap-free block writes --- libraries/tracereader/tracefiles.go | 21 +-- .../tracereader/tracefiles_config_test.go | 84 ++++++++++++ services/coreindex/cmd/coreindex/main.go | 121 ++++++++++++------ services/coreindex/cmd/coreindex/readclass.go | 89 +++++++++++++ .../coreindex/cmd/coreindex/readclass_test.go | 58 +++++++++ .../internal/appendlog/slice_store.go | 14 ++ .../internal/appendlog/slice_store_test.go | 65 ++++++++++ 7 files changed, 408 insertions(+), 44 deletions(-) create mode 100644 services/coreindex/cmd/coreindex/readclass.go create mode 100644 services/coreindex/cmd/coreindex/readclass_test.go diff --git a/libraries/tracereader/tracefiles.go b/libraries/tracereader/tracefiles.go index 28d158e..2882c75 100644 --- a/libraries/tracereader/tracefiles.go +++ b/libraries/tracereader/tracefiles.go @@ -17,6 +17,7 @@ import ( const HeaderVersion uint32 = 1 var ErrNotFound = errors.New("not found") +var ErrIncompleteData = errors.New("incomplete trace data") type Config struct { Debug bool @@ -160,7 +161,7 @@ func GetRawBlocksWithMetadata(blockNum uint32, limit int, conf *Config) ([]RawBl for { err = dec.DecodeVariant(&entry) if err != nil { - if err == io.EOF { + if err == io.EOF || errors.Is(err, io.ErrUnexpectedEOF) { break } return nil, err @@ -206,14 +207,14 @@ func GetRawBlocksWithMetadata(blockNum uint32, limit int, conf *Config) ([]RawBl } if int64(offset) >= traceFileSize { - return nil, fmt.Errorf("block %d: offset %d exceeds trace file size %d (stride=%s)", - bn, offset, traceFileSize, stride) + return nil, fmt.Errorf("block %d: offset %d exceeds trace file size %d (stride=%s): %w", + bn, offset, traceFileSize, stride, ErrIncompleteData) } 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)", - bn, blockSize, offset, actualAvailable, traceFileSize, stride) + return nil, fmt.Errorf("block %d: expected %d bytes at offset %d but only %d bytes available in trace file (size=%d, stride=%s): %w", + bn, blockSize, offset, actualAvailable, traceFileSize, stride, ErrIncompleteData) } if _, err := slice.Seek(int64(offset), io.SeekStart); err != nil { @@ -226,8 +227,8 @@ func GetRawBlocksWithMetadata(blockNum uint32, limit int, conf *Config) ([]RawBl 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)", - bn, n, blockSize, offset, traceFileSize, stride) + return nil, fmt.Errorf("block %d: read %d bytes but expected %d (offset=%d, fileSize=%d, stride=%s): %w", + bn, n, blockSize, offset, traceFileSize, stride, ErrIncompleteData) } rawBlocks = append(rawBlocks, RawBlockData{ @@ -367,7 +368,7 @@ func GetInfo(conf *Config) (GetInfoResponse, error) { for { err = dec.DecodeVariant(&entry) if err != nil { - if err == io.EOF { + if err == io.EOF || errors.Is(err, io.ErrUnexpectedEOF) { break } return rv, err @@ -403,6 +404,10 @@ func openSliceIndex(filename string, conf *Config) (*os.File, error) { var version uint32 err = binary.Read(f, binary.LittleEndian, &version) if err != nil { + if err == io.EOF || errors.Is(err, io.ErrUnexpectedEOF) { + f.Close() + return nil, ErrNotFound + } return nil, err } diff --git a/libraries/tracereader/tracefiles_config_test.go b/libraries/tracereader/tracefiles_config_test.go index 702742e..e0ae977 100644 --- a/libraries/tracereader/tracefiles_config_test.go +++ b/libraries/tracereader/tracefiles_config_test.go @@ -3,6 +3,7 @@ package tracereader import ( "bytes" "encoding/binary" + "errors" "os" "path/filepath" "testing" @@ -453,6 +454,58 @@ func createTestIndexFileWithMultipleBlocks(path string, startBlock, endBlock uin return nil } +func createTestIndexFileWithPartialTail(path string, startBlock, endBlock uint32) error { + f, err := os.Create(path) + if err != nil { + return err + } + defer f.Close() + + if err := binary.Write(f, binary.LittleEndian, HeaderVersion); err != nil { + return err + } + + offset := uint64(0) + blockDataSize := uint64(100) + for blockNum := startBlock; blockNum <= endBlock; blockNum++ { + f.Write([]byte{0}) + + id := [32]byte{byte(blockNum & 0xFF), byte((blockNum >> 8) & 0xFF)} + f.Write(id[:]) + binary.Write(f, binary.LittleEndian, blockNum) + binary.Write(f, binary.LittleEndian, offset) + offset += blockDataSize + } + + f.Write([]byte{0}) + f.Write([]byte{0xAA, 0xBB, 0xCC}) + + return nil +} + +func TestGetRawBlocksWithMetadata_PartialTrailingRecord(t *testing.T) { + tmpDir := t.TempDir() + conf := &Config{Debug: false, Stride: 500, Dir: tmpDir} + + indexFile := filepath.Join(tmpDir, "trace_index_0000000000-0000000500.log") + if err := createTestIndexFileWithPartialTail(indexFile, 100, 105); err != nil { + t.Fatalf("Failed to create test index file: %v", err) + } + + traceFile := filepath.Join(tmpDir, "trace_0000000000-0000000500.log") + if err := os.WriteFile(traceFile, make([]byte, 600), 0644); err != nil { + t.Fatalf("Failed to create trace file: %v", err) + } + + rawBlocks, err := GetRawBlocksWithMetadata(100, 1, conf) + if err != nil { + t.Fatalf("GetRawBlocksWithMetadata() with partial trailing record = %v; want success (block 100's entry is complete)", err) + } + if len(rawBlocks) != 1 || rawBlocks[0].BlockNum != 100 { + t.Fatalf("got %d blocks (%v); want exactly block 100", len(rawBlocks), rawBlocks) + } +} + func TestGetInfo_DebugMode(t *testing.T) { tmpDir := t.TempDir() conf := &Config{ @@ -519,4 +572,35 @@ func TestOpenSliceIndex_EmptyFile(t *testing.T) { } } +func TestGetRawBlocksWithMetadata_IncompleteData(t *testing.T) { + tmpDir := t.TempDir() + conf := &Config{Debug: false, Stride: 500, Dir: tmpDir} + + indexFile := filepath.Join(tmpDir, "trace_index_0000000000-0000000500.log") + if err := createTestIndexFileWithMultipleBlocks(indexFile, 100, 105); err != nil { + t.Fatalf("Failed to create test index file: %v", err) + } + traceFile := filepath.Join(tmpDir, "trace_0000000000-0000000500.log") + if err := os.WriteFile(traceFile, make([]byte, 50), 0644); err != nil { + t.Fatalf("Failed to create trace file: %v", err) + } + + _, err := GetRawBlocksWithMetadata(100, 1, conf) + if !errors.Is(err, ErrIncompleteData) { + t.Fatalf("expected ErrIncompleteData, got %v", err) + } +} + +func TestOpenSliceIndex_PartialHeader(t *testing.T) { + tmpDir := t.TempDir() + indexFile := filepath.Join(tmpDir, "trace_index_partial.log") + if err := os.WriteFile(indexFile, []byte{0x01, 0x00}, 0644); err != nil { + t.Fatalf("write: %v", err) + } + conf := &Config{Debug: false} + if _, err := openSliceIndex(indexFile, conf); !errors.Is(err, ErrNotFound) { + t.Fatalf("expected ErrNotFound for partial header, got %v", err) + } +} + // ============================================================================= diff --git a/services/coreindex/cmd/coreindex/main.go b/services/coreindex/cmd/coreindex/main.go index 5ad69ea..2f066f3 100644 --- a/services/coreindex/cmd/coreindex/main.go +++ b/services/coreindex/cmd/coreindex/main.go @@ -6,8 +6,8 @@ import ( "encoding/hex" "encoding/json" "errors" - "log" "fmt" + "log" "net/http" _ "net/http/pprof" "os" @@ -136,11 +136,13 @@ type writeBatch struct { } type processedStride struct { - strideNum uint32 - strideStart uint32 - strideEnd uint32 - blocks []appendlog.BlockEntry // All blocks in stride, in order - abis []abiEntry // ABIs extracted from setabi actions + strideNum uint32 + strideStart uint32 + strideEnd uint32 + blocks []appendlog.BlockEntry + abis []abiEntry + incompleteFrom uint32 + readErr error } type abiEntry struct { @@ -561,6 +563,9 @@ func isSliceError(err error) bool { if err == nil { return false } + if errors.Is(err, appendlog.ErrNonContiguousWrite) { + return false + } errStr := err.Error() slicePatterns := []string{ "block count mismatch", @@ -671,6 +676,7 @@ func syncFromTraceFiles(config *server.Config, traceConfig *tracereader.Config, batch := make([]appendlog.BlockEntry, 0, batchSize) var sliceErr error + var readErr error flushBatch := func() error { if len(batch) == 0 { @@ -789,6 +795,14 @@ func syncFromTraceFiles(config *server.Config, traceConfig *tracereader.Config, return nil } + classifyTip := func(ps *processedStride) bool { + if ps.incompleteFrom != 0 || ps.readErr != nil { + readErr = classifyStrideTail(ps, end) + return true + } + return false + } + // Main loop: receive strides, reorder, write blocks mainLoop: for !(*exit) && nextStrideNum <= endStride { @@ -804,11 +818,17 @@ mainLoop: logger.Error("Failed to write stride %d: %v", nextStrideNum, err) if isSliceError(err) { sliceErr = err + } else { + readErr = err } *exit = true break mainLoop } + if classifyTip(ps) { + break mainLoop + } + nextStrideNum++ } @@ -897,7 +917,7 @@ mainLoop: } // Process any remaining pending strides in order - for nextStrideNum <= endStride && !(*exit) { + for nextStrideNum <= endStride && !(*exit) && readErr == nil { ps, ok := pendingStrides[nextStrideNum] if !ok { // Stride not available yet - drain from channel @@ -917,9 +937,14 @@ mainLoop: logger.Error("Failed to write stride %d: %v", nextStrideNum, err) if isSliceError(err) { sliceErr = err + } else { + readErr = err } break } + if classifyTip(ps) { + break + } nextStrideNum++ } @@ -958,6 +983,9 @@ mainLoop: logger.Println("debug-shutdown", "Async writer finished") } + if readErr != nil { + return readErr + } return sliceErr } @@ -1119,10 +1147,19 @@ func gophers(config *server.Config, traceConfig *tracereader.Config, strideChan break } - // Check for I/O errors if prefetched.err != nil { - logger.Fatal("Failed to read trace files for blocks %d-%d: %v (sync requires complete trace files)", - prefetched.strideStart, prefetched.strideEnd, prefetched.err) + select { + case strideChan <- &processedStride{ + strideNum: prefetched.strideNum, + strideStart: prefetched.strideStart, + strideEnd: prefetched.strideEnd, + incompleteFrom: prefetched.strideStart, + readErr: prefetched.err, + }: + case <-exitChan: + return + } + continue } // Track stride being processed by this worker @@ -1153,6 +1190,7 @@ func gophers(config *server.Config, traceConfig *tracereader.Config, strideChan // Accumulate all blocks in stride for single output strideBlocks := make([]appendlog.BlockEntry, 0, expectedBlocks) + incompleteFrom := uint32(0) var strideABIs []abiEntry exitRequested := false @@ -1163,14 +1201,9 @@ func gophers(config *server.Config, traceConfig *tracereader.Config, strideChan break } - // Check if this block has trace data available if idx >= len(rawBlocks) { - // Block is missing - append marker entry (nil data) - strideBlocks = append(strideBlocks, appendlog.BlockEntry{ - BlockNum: blockNum, - Data: nil, // Marker for missing block - }) - continue + incompleteFrom = blockNum + break } // INSTRUMENTATION: Time block processing @@ -1286,16 +1319,16 @@ func gophers(config *server.Config, traceConfig *tracereader.Config, strideChan atomic.AddInt64(&workerStrideTimeNs[i], time.Since(strideProcessStart).Nanoseconds()) } - // Send completed stride (unless exit requested) - if !exitRequested && len(strideBlocks) > 0 { + if !exitRequested && (len(strideBlocks) > 0 || incompleteFrom != 0) { t2 := time.Now() select { case strideChan <- &processedStride{ - strideNum: prefetched.strideNum, - strideStart: strideStart, - strideEnd: strideEnd, - blocks: strideBlocks, - abis: strideABIs, + strideNum: prefetched.strideNum, + strideStart: strideStart, + strideEnd: strideEnd, + blocks: strideBlocks, + abis: strideABIs, + incompleteFrom: incompleteFrom, }: atomic.AddInt64(&totalChanSendNs, time.Since(t2).Nanoseconds()) case <-exitChan: @@ -2399,10 +2432,11 @@ func main() { // Check for new blocks every 500ms (EOS block interval) syncWaitStart := time.Now() liveSyncAnnounced := false // Track if we've announced live sync mode - monitoringModeAnnounced := false // Track if we've announced monitoring mode - lastMonitoringLog := time.Time{} // Track last monitoring heartbeat - lastStoreAheadLog := time.Time{} // Track last "store ahead" warning - lastMediumBatchLog := time.Time{} // Track last medium batch log (rate limit) + var stalls stallTracker + monitoringModeAnnounced := false // Track if we've announced monitoring mode + lastMonitoringLog := time.Time{} // Track last monitoring heartbeat + lastStoreAheadLog := time.Time{} // Track last "store ahead" warning + lastMediumBatchLog := time.Time{} // Track last medium batch log (rate limit) // Periodic slice cache save (every 60 seconds if slice count changed) lastSavedSliceCount := len(store.GetSliceInfos()) go func() { @@ -2518,15 +2552,30 @@ func main() { } syncErr := syncFromTraceFiles(serverConfig, traceConfig, uint32(cfg.Workers), uint32(cfg.Prefetchers), store, abiWriter, myLIB+1, syncTarget, &exit, ctx.Done(), broadcastFn, cfg.GetLogInterval()) - // Handle slice errors with reactive validation - if syncErr != nil && !exit { - logger.Warning("Sync error detected: %v", syncErr) - logger.Println("validation", "Running reactive validation to check/repair slices...") - if repairErr := store.ValidateAndRepairLastSlices(2); repairErr != nil { - logger.Error("Reactive validation failed: %v", repairErr) - logger.Fatal("Cannot continue after slice validation failure") + switch e := syncErr.(type) { + case nil: + stalls.reset() + case *transientReadError: + now := time.Now() + if stalls.onTransient(e.block, now, stallEscalationThreshold*time.Second) { + logger.Error("Sync stalled at block %d for %s — nodeos may be down (still retrying)", + e.block, stalls.stalledFor(now).Round(time.Second)) + } + case *fatalReadError: + logger.Fatal("%v", e) + default: + if errors.Is(syncErr, appendlog.ErrNonContiguousWrite) { + logger.Fatal("Gap-safety violation (non-contiguous write): %v", syncErr) + } + if !exit { + logger.Warning("Sync error detected: %v", syncErr) + logger.Println("validation", "Running reactive validation to check/repair slices...") + if repairErr := store.ValidateAndRepairLastSlices(2); repairErr != nil { + logger.Error("Reactive validation failed: %v", repairErr) + logger.Fatal("Cannot continue after slice validation failure") + } + logger.Println("validation", "Reactive validation complete, will restart sync from repaired position") } - logger.Println("validation", "Reactive validation complete, will restart sync from repaired position") } if exit { diff --git a/services/coreindex/cmd/coreindex/readclass.go b/services/coreindex/cmd/coreindex/readclass.go new file mode 100644 index 0000000..fde13d7 --- /dev/null +++ b/services/coreindex/cmd/coreindex/readclass.go @@ -0,0 +1,89 @@ +package main + +import ( + "errors" + "fmt" + "time" + + "github.com/greymass/roborovski/libraries/tracereader" +) + +const liveEdgeWindow uint32 = 400 + +const stallEscalationThreshold = 30 + +type readErrorClass int + +const ( + readFatal readErrorClass = iota + readTransient +) + +func isAvailabilityError(err error) bool { + return err != nil && + (errors.Is(err, tracereader.ErrNotFound) || errors.Is(err, tracereader.ErrIncompleteData)) +} + +func classifyReadError(err error, failBlock, end uint32) readErrorClass { + if err != nil && !isAvailabilityError(err) { + return readFatal + } + if failBlock+liveEdgeWindow >= end { + return readTransient + } + return readFatal +} + +type transientReadError struct{ block uint32 } + +func (e *transientReadError) Error() string { + return fmt.Sprintf("transient trace read: block %d not yet durable at tip", e.block) +} + +type fatalReadError struct { + block uint32 + cause error +} + +func (e *fatalReadError) Error() string { + if e.cause != nil { + return fmt.Sprintf("unrecoverable trace read at block %d: %v", e.block, e.cause) + } + return fmt.Sprintf("missing block %d in finalized history (gap)", e.block) +} + +type stallTracker struct { + block uint32 + since time.Time + escalated bool +} + +func (st *stallTracker) onTransient(block uint32, now time.Time, threshold time.Duration) bool { + if block != st.block { + st.block = block + st.since = now + st.escalated = false + return false + } + if !st.escalated && now.Sub(st.since) >= threshold { + st.escalated = true + return true + } + return false +} + +func (st *stallTracker) reset() { *st = stallTracker{} } + +func (st *stallTracker) stalledFor(now time.Time) time.Duration { + if st.block == 0 { + return 0 + } + return now.Sub(st.since) +} + +func classifyStrideTail(ps *processedStride, end uint32) error { + if classifyReadError(ps.readErr, ps.incompleteFrom, end) == readTransient { + return &transientReadError{block: ps.incompleteFrom} + } + return &fatalReadError{block: ps.incompleteFrom, cause: ps.readErr} +} diff --git a/services/coreindex/cmd/coreindex/readclass_test.go b/services/coreindex/cmd/coreindex/readclass_test.go new file mode 100644 index 0000000..ad084da --- /dev/null +++ b/services/coreindex/cmd/coreindex/readclass_test.go @@ -0,0 +1,58 @@ +package main + +import ( + "fmt" + "io" + "testing" + "time" + + "github.com/greymass/roborovski/libraries/tracereader" +) + +func TestClassifyReadError(t *testing.T) { + const end = 1000 + cases := []struct { + name string + err error + failBlock uint32 + want readErrorClass + }{ + {"short stride at tip", nil, 1000, readTransient}, + {"short stride just inside window", nil, end - liveEdgeWindow, readTransient}, + {"short stride deep in history", nil, 10, readFatal}, + {"availability err at tip", tracereader.ErrNotFound, 999, readTransient}, + {"incomplete-data err at tip", fmt.Errorf("x: %w", tracereader.ErrIncompleteData), 999, readTransient}, + {"availability err deep", tracereader.ErrNotFound, 5, readFatal}, + {"os error at tip is fatal", io.ErrClosedPipe, 1000, readFatal}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := classifyReadError(c.err, c.failBlock, end); got != c.want { + t.Fatalf("classifyReadError = %v, want %v", got, c.want) + } + }) + } +} + +func TestStallTracker(t *testing.T) { + base := time.Unix(1_700_000_000, 0) + var st stallTracker + thr := 30 * time.Second + + if st.onTransient(900, base, thr) { + t.Fatal("should not escalate on first sighting") + } + if st.onTransient(900, base.Add(10*time.Second), thr) { + t.Fatal("should not escalate before threshold") + } + if !st.onTransient(900, base.Add(31*time.Second), thr) { + t.Fatal("should escalate after threshold") + } + if st.onTransient(901, base.Add(40*time.Second), thr) { + t.Fatal("advancing to a new block should reset the stall timer") + } + st.reset() + if st.onTransient(901, base.Add(100*time.Second), thr) { + t.Fatal("after reset, first sighting should not escalate") + } +} diff --git a/services/coreindex/internal/appendlog/slice_store.go b/services/coreindex/internal/appendlog/slice_store.go index b974537..0303299 100644 --- a/services/coreindex/internal/appendlog/slice_store.go +++ b/services/coreindex/internal/appendlog/slice_store.go @@ -13,6 +13,8 @@ import ( "github.com/greymass/roborovski/libraries/logger" ) +var ErrNonContiguousWrite = errors.New("non-contiguous block write") + // SliceStore manages append-only log files with tiered slice-based storage // This is the production implementation for large datasets (400M+ blocks) // Memory usage stays bounded regardless of dataset size. @@ -199,6 +201,10 @@ func (s *SliceStore) AppendBlock(blockNum uint32, data []byte, globMin, globMax s.mu.Lock() defer s.mu.Unlock() + if s.head != 0 && blockNum != s.head+1 { + return fmt.Errorf("%w: expected block %d, got %d", ErrNonContiguousWrite, s.head+1, blockNum) + } + // Compress data if enabled dataToWrite := data if s.enableZstd { @@ -573,6 +579,14 @@ func (s *SliceStore) AppendBlockBatch(blocks []BlockEntry) error { lockWaitTime := time.Since(batchStart) defer s.mu.Unlock() + prev := s.head + for i := range blocks { + if prev != 0 && blocks[i].BlockNum != prev+1 { + return fmt.Errorf("%w: expected block %d, got %d", ErrNonContiguousWrite, prev+1, blocks[i].BlockNum) + } + prev = blocks[i].BlockNum + } + // Process blocks sequentially, handling rotations as needed var currentSlice *Slice var err error diff --git a/services/coreindex/internal/appendlog/slice_store_test.go b/services/coreindex/internal/appendlog/slice_store_test.go index 008bef0..d264664 100644 --- a/services/coreindex/internal/appendlog/slice_store_test.go +++ b/services/coreindex/internal/appendlog/slice_store_test.go @@ -1,6 +1,7 @@ package appendlog import ( + "errors" "os" "path/filepath" "testing" @@ -652,3 +653,67 @@ func TestSliceStore_DirectoryStructure(t *testing.T) { } // Note: globalseq.index is no longer created (removed Jan 2025 for performance) } + +func newTestStore(t *testing.T) *SliceStore { + t.Helper() + store, err := NewSliceStore(t.TempDir(), SliceStoreOptions{ + BlocksPerSlice: 100, + BlockCacheSize: 10, + Debug: false, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { store.Close() }) + return store +} + +func TestAppendBlock_RejectsNonContiguous(t *testing.T) { + store := newTestStore(t) + data := []byte("x") + + if err := store.AppendBlock(2, data, 1, 1, false); err != nil { + t.Fatalf("first append: %v", err) + } + err := store.AppendBlock(4, data, 2, 2, false) + if !errors.Is(err, ErrNonContiguousWrite) { + t.Fatalf("expected ErrNonContiguousWrite, got %v", err) + } + if store.GetHead() != 2 { + t.Fatalf("head must be unchanged after rejected write, got %d", store.GetHead()) + } + if err := store.AppendBlock(3, data, 2, 2, false); err != nil { + t.Fatalf("contiguous append: %v", err) + } +} + +func TestAppendBlockBatch_RejectsInternalGap(t *testing.T) { + store := newTestStore(t) + d := []byte("x") + batch := []BlockEntry{ + {BlockNum: 2, Data: d, GlobMin: 1, GlobMax: 1}, + {BlockNum: 4, Data: d, GlobMin: 2, GlobMax: 2}, + } + if err := store.AppendBlockBatch(batch); !errors.Is(err, ErrNonContiguousWrite) { + t.Fatalf("expected ErrNonContiguousWrite, got %v", err) + } + if store.GetHead() != 0 { + t.Fatalf("head must be unchanged after rejected batch, got %d", store.GetHead()) + } +} + +func TestAppendBlockBatch_AcceptsContiguous(t *testing.T) { + store := newTestStore(t) + d := []byte("x") + batch := []BlockEntry{ + {BlockNum: 2, Data: d, GlobMin: 1, GlobMax: 1}, + {BlockNum: 3, Data: d, GlobMin: 2, GlobMax: 2}, + {BlockNum: 4, Data: d, GlobMin: 3, GlobMax: 3}, + } + if err := store.AppendBlockBatch(batch); err != nil { + t.Fatalf("contiguous batch: %v", err) + } + if store.GetHead() != 4 { + t.Fatalf("expected head 4, got %d", store.GetHead()) + } +} From 9728b81ed53172ba9c943d8a800b306aede34cf3 Mon Sep 17 00:00:00 2001 From: aaroncox Date: Sat, 6 Jun 2026 14:03:58 -0700 Subject: [PATCH 2/2] stream action ordinals end-to-end; fix authorizer-indexed catchup, speculative block selection, broadcast --- libraries/actionstream/client.go | 65 +++++++++++-------- libraries/actionstream/client_test.go | 51 +++++++++++++-- libraries/corereader/canonical_filter.go | 55 ++++++---------- libraries/corereader/canonical_filter_test.go | 43 ++++++++++++ libraries/corereader/shared.go | 4 +- libraries/corereader/slice_reader.go | 24 +++---- libraries/corereader/types.go | 18 +++-- services/actionindex/internal/broadcaster.go | 21 +++--- .../actionindex/internal/stream_catchup.go | 19 +++--- .../internal/stream_catchup_test.go | 43 +++++++++++- services/actionindex/internal/stream_tcp.go | 7 +- .../actionindex/internal/stream_tcp_test.go | 60 +++++++++++++++++ services/actionindex/internal/sync.go | 21 +++--- .../internal/sync_broadcast_test.go | 50 ++++++++++++++ 14 files changed, 360 insertions(+), 121 deletions(-) create mode 100644 services/actionindex/internal/stream_tcp_test.go diff --git a/libraries/actionstream/client.go b/libraries/actionstream/client.go index 9649401..6d50bf0 100644 --- a/libraries/actionstream/client.go +++ b/libraries/actionstream/client.go @@ -20,13 +20,13 @@ var ( ) const ( - MsgTypeActionSubscribe uint8 = 0x30 - MsgTypeActionAck uint8 = 0x31 - MsgTypeActionBatch uint8 = 0x32 - MsgTypeActionHeartbeat uint8 = 0x33 - MsgTypeActionError uint8 = 0x34 - MsgTypeActionDecoded uint8 = 0x35 - MsgTypeCatchupComplete uint8 = 0x36 + MsgTypeActionSubscribe uint8 = 0x30 + MsgTypeActionAck uint8 = 0x31 + MsgTypeActionBatch uint8 = 0x32 + MsgTypeActionHeartbeat uint8 = 0x33 + MsgTypeActionError uint8 = 0x34 + MsgTypeActionDecoded uint8 = 0x35 + MsgTypeCatchupComplete uint8 = 0x36 MaxMessageSize = 10 * 1024 * 1024 ) @@ -51,15 +51,18 @@ func DefaultClientConfig() ClientConfig { } type Action struct { - GlobalSeq uint64 - BlockNum uint32 - BlockTime uint32 - Contract string - Action string - Receiver string - ActionData []byte - CpuUsageUs uint32 - NetUsageWords uint32 + GlobalSeq uint64 + BlockNum uint32 + BlockTime uint32 + Contract string + Action string + Receiver string + ActionData []byte + CpuUsageUs uint32 + NetUsageWords uint32 + ActionOrdinal uint32 + CreatorActionOrdinal uint32 + ClosestUnnotifiedAncestorActionOrdinal uint32 } type Filter struct { @@ -492,7 +495,7 @@ func (c *Client) writeMessage(w io.Writer, msgType uint8, payload []byte) error } func (c *Client) decodeAction(payload []byte) (Action, error) { - if len(payload) < 48 { + if len(payload) < 60 { return Action{}, errors.New("action payload too short") } @@ -504,22 +507,28 @@ func (c *Client) decodeAction(payload []byte) (Action, error) { receiver := binary.LittleEndian.Uint64(payload[32:40]) cpuUsageUs := binary.LittleEndian.Uint32(payload[40:44]) netUsageWords := binary.LittleEndian.Uint32(payload[44:48]) + actionOrdinal := binary.LittleEndian.Uint32(payload[48:52]) + creatorActionOrdinal := binary.LittleEndian.Uint32(payload[52:56]) + closestUAAO := binary.LittleEndian.Uint32(payload[56:60]) var actionData []byte - if len(payload) > 48 { - actionData = payload[48:] + if len(payload) > 60 { + actionData = payload[60:] } return Action{ - GlobalSeq: globalSeq, - BlockNum: blockNum, - BlockTime: blockTime, - Contract: chain.NameToString(contract), - Action: chain.NameToString(actionName), - Receiver: chain.NameToString(receiver), - ActionData: actionData, - CpuUsageUs: cpuUsageUs, - NetUsageWords: netUsageWords, + GlobalSeq: globalSeq, + BlockNum: blockNum, + BlockTime: blockTime, + Contract: chain.NameToString(contract), + Action: chain.NameToString(actionName), + Receiver: chain.NameToString(receiver), + ActionData: actionData, + CpuUsageUs: cpuUsageUs, + NetUsageWords: netUsageWords, + ActionOrdinal: actionOrdinal, + CreatorActionOrdinal: creatorActionOrdinal, + ClosestUnnotifiedAncestorActionOrdinal: closestUAAO, }, nil } diff --git a/libraries/actionstream/client_test.go b/libraries/actionstream/client_test.go index 37e8dc6..f4807e2 100644 --- a/libraries/actionstream/client_test.go +++ b/libraries/actionstream/client_test.go @@ -28,7 +28,7 @@ func TestNewClient(t *testing.T) { func TestDecodeAction(t *testing.T) { client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig()) - payload := make([]byte, 48) + payload := make([]byte, 60) binary.LittleEndian.PutUint64(payload[0:8], 12345) binary.LittleEndian.PutUint32(payload[8:12], 1000) binary.LittleEndian.PutUint32(payload[12:16], 1700000000) @@ -37,6 +37,9 @@ func TestDecodeAction(t *testing.T) { binary.LittleEndian.PutUint64(payload[32:40], 0x5530EA033C80A555) binary.LittleEndian.PutUint32(payload[40:44], 500) binary.LittleEndian.PutUint32(payload[44:48], 20) + binary.LittleEndian.PutUint32(payload[48:52], 0) + binary.LittleEndian.PutUint32(payload[52:56], 0) + binary.LittleEndian.PutUint32(payload[56:60], 0) action, err := client.decodeAction(payload) if err != nil { @@ -63,7 +66,7 @@ func TestDecodeAction(t *testing.T) { func TestDecodeActionWithData(t *testing.T) { client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig()) - payload := make([]byte, 56) + payload := make([]byte, 68) binary.LittleEndian.PutUint64(payload[0:8], 100) binary.LittleEndian.PutUint32(payload[8:12], 500) binary.LittleEndian.PutUint32(payload[12:16], 1600000000) @@ -72,7 +75,10 @@ func TestDecodeActionWithData(t *testing.T) { binary.LittleEndian.PutUint64(payload[32:40], 0) binary.LittleEndian.PutUint32(payload[40:44], 0) binary.LittleEndian.PutUint32(payload[44:48], 0) - copy(payload[48:], []byte("testdata")) + binary.LittleEndian.PutUint32(payload[48:52], 0) + binary.LittleEndian.PutUint32(payload[52:56], 0) + binary.LittleEndian.PutUint32(payload[56:60], 0) + copy(payload[60:], []byte("testdata")) action, err := client.decodeAction(payload) if err != nil { @@ -87,10 +93,41 @@ func TestDecodeActionWithData(t *testing.T) { func TestDecodeActionTooShort(t *testing.T) { client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig()) - payload := make([]byte, 30) + payload := make([]byte, 48) _, err := client.decodeAction(payload) if err == nil { - t.Error("expected error for short payload") + t.Error("expected error for payload shorter than 60 bytes") + } +} + +func TestDecodeActionOrdinals(t *testing.T) { + client := NewClient("localhost:9411", Filter{}, 0, DefaultClientConfig()) + + build := func(ao, creator, closest uint32) []byte { + p := make([]byte, 60) + binary.LittleEndian.PutUint64(p[0:8], 7) + binary.LittleEndian.PutUint32(p[48:52], ao) + binary.LittleEndian.PutUint32(p[52:56], creator) + binary.LittleEndian.PutUint32(p[56:60], closest) + return p + } + + topLevel, err := client.decodeAction(build(1, 0, 0)) + if err != nil { + t.Fatalf("decodeAction (top-level) failed: %v", err) + } + if topLevel.ActionOrdinal != 1 || topLevel.CreatorActionOrdinal != 0 || topLevel.ClosestUnnotifiedAncestorActionOrdinal != 0 { + t.Errorf("top-level ordinals = (%d,%d,%d), want (1,0,0)", + topLevel.ActionOrdinal, topLevel.CreatorActionOrdinal, topLevel.ClosestUnnotifiedAncestorActionOrdinal) + } + + inline, err := client.decodeAction(build(3, 1, 1)) + if err != nil { + t.Fatalf("decodeAction (inline) failed: %v", err) + } + if inline.ActionOrdinal != 3 || inline.CreatorActionOrdinal != 1 || inline.ClosestUnnotifiedAncestorActionOrdinal != 1 { + t.Errorf("inline ordinals = (%d,%d,%d), want (3,1,1)", + inline.ActionOrdinal, inline.CreatorActionOrdinal, inline.ClosestUnnotifiedAncestorActionOrdinal) } } @@ -644,7 +681,7 @@ func TestDialReceivesBatch(t *testing.T) { payload := make([]byte, length-1) conn.Read(payload) - actionPayload := make([]byte, 48) + actionPayload := make([]byte, 60) binary.LittleEndian.PutUint64(actionPayload[0:8], 999) binary.LittleEndian.PutUint32(actionPayload[8:12], 100) binary.LittleEndian.PutUint32(actionPayload[12:16], 1700000000) @@ -832,7 +869,7 @@ func TestRecvLoopReceivesActions(t *testing.T) { close(serverReady) for i := uint64(0); i < 3; i++ { - actionPayload := make([]byte, 48) + actionPayload := make([]byte, 60) 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/corereader/canonical_filter.go b/libraries/corereader/canonical_filter.go index 2afa45e..f8c5f3f 100644 --- a/libraries/corereader/canonical_filter.go +++ b/libraries/corereader/canonical_filter.go @@ -77,6 +77,21 @@ type combinedActionInfo struct { TrxIndex uint32 } +func newFilteredAction(account uint64, info *combinedActionInfo, globalSeq uint64, isAuth bool) Action { + return Action{ + Account: account, + Contract: info.Meta.Contract, + Action: info.Meta.Action, + GlobalSeq: globalSeq, + TrxIndex: info.TrxIndex, + IsAuthorizer: isAuth, + Receiver: info.Receiver, + ActionOrdinal: info.Action.ActionOrdinal, + CreatorAO: info.Action.CreatorAO, + ClosestUAAO: info.Action.ClosestUAAO, + } +} + type blockFilter struct { // Slice-backed lookup: avoids struct copy on every map access infoSlice []combinedActionInfo @@ -315,15 +330,7 @@ func filterBlockInto(bf *blockFilter, notif RawBlock, actionFilter ActionFilterF } } } - actionsBuf = append(actionsBuf, Action{ - Account: account, - Contract: info.Meta.Contract, - Action: info.Meta.Action, - GlobalSeq: globalSeq, - TrxIndex: info.TrxIndex, - IsAuthorizer: isAuth, - Receiver: info.Receiver, - }) + actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, isAuth)) if minSeq == 0 || globalSeq < minSeq { minSeq = globalSeq } @@ -371,15 +378,7 @@ func filterBlockInto(bf *blockFilter, notif RawBlock, actionFilter ActionFilterF } } - actionsBuf = append(actionsBuf, Action{ - Account: account, - Contract: info.Meta.Contract, - Action: info.Meta.Action, - GlobalSeq: globalSeq, - TrxIndex: info.TrxIndex, - IsAuthorizer: true, - Receiver: info.Receiver, - }) + actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, true)) if minSeq == 0 || globalSeq < minSeq { minSeq = globalSeq } @@ -499,15 +498,7 @@ func filterBlockIntoTimed(bf *blockFilter, notif RawBlock, actionFilter ActionFi } } } - actionsBuf = append(actionsBuf, Action{ - Account: account, - Contract: info.Meta.Contract, - Action: info.Meta.Action, - GlobalSeq: globalSeq, - TrxIndex: info.TrxIndex, - IsAuthorizer: isAuth, - Receiver: info.Receiver, - }) + actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, isAuth)) if minSeq == 0 || globalSeq < minSeq { minSeq = globalSeq } @@ -558,15 +549,7 @@ func filterBlockIntoTimed(bf *blockFilter, notif RawBlock, actionFilter ActionFi } } - actionsBuf = append(actionsBuf, Action{ - Account: account, - Contract: info.Meta.Contract, - Action: info.Meta.Action, - GlobalSeq: globalSeq, - TrxIndex: info.TrxIndex, - IsAuthorizer: true, - Receiver: info.Receiver, - }) + actionsBuf = append(actionsBuf, newFilteredAction(account, info, globalSeq, true)) if minSeq == 0 || globalSeq < minSeq { minSeq = globalSeq } diff --git a/libraries/corereader/canonical_filter_test.go b/libraries/corereader/canonical_filter_test.go index 8f9735f..9e5ec56 100644 --- a/libraries/corereader/canonical_filter_test.go +++ b/libraries/corereader/canonical_filter_test.go @@ -4,6 +4,49 @@ import ( "testing" ) +func TestFilterRawBlock_CarriesOrdinals(t *testing.T) { + namesInBlock := []uint64{1111, 5555} + + notif := RawBlock{ + BlockNum: 100, + BlockTime: 1000000, + NamesInBlock: namesInBlock, + Notifications: map[uint64][]uint64{ + 1111: {1}, + }, + ActionMeta: []ActionMetadata{ + {GlobalSeq: 1, Contract: 5555, Action: 6666}, + }, + Actions: []CanonicalAction{ + { + ActionOrdinal: 3, + CreatorAO: 1, + ClosestUAAO: 1, + ReceiverUint64: 5555, + DataIndex: 0, + AuthAccountIndexes: []uint32{0}, + GlobalSeqUint64: 1, + TrxIndex: 0, + ContractUint64: 5555, + ActionUint64: 6666, + }, + }, + } + + filtered := FilterRawBlock(notif, nil) + if len(filtered.Actions) == 0 { + t.Fatal("expected filtered actions, got 0") + } + for _, a := range filtered.Actions { + if a.GlobalSeq != 1 { + continue + } + if a.ActionOrdinal != 3 || a.CreatorAO != 1 || a.ClosestUAAO != 1 { + t.Errorf("seq 1 ordinals = (%d,%d,%d), want (3,1,1)", a.ActionOrdinal, a.CreatorAO, a.ClosestUAAO) + } + } +} + func TestFilterRawBlock_Actions(t *testing.T) { namesInBlock := []uint64{1111, 2222, 5555, 8888} diff --git a/libraries/corereader/shared.go b/libraries/corereader/shared.go index 5bea3db..850f621 100644 --- a/libraries/corereader/shared.go +++ b/libraries/corereader/shared.go @@ -220,6 +220,7 @@ type syncBlockData struct { type syncActionInfo struct { ActionOrdinal uint32 CreatorAO uint32 + ClosestUAAO uint32 ReceiverIndex uint32 ContractNameIndex uint32 ActionNameIndex uint32 @@ -537,8 +538,9 @@ func parseSyncActionDirect(r *byteSliceReader, blob *syncBlockBlob, authBufOffse action.CreatorAO = uint32(r.ReadUvarint()) } + action.ClosestUAAO = action.CreatorAO if (magicmask1 & (1 << 3)) == 0 { - r.ReadUvarint() + action.ClosestUAAO = action.CreatorAO ^ uint32(r.ReadUvarint()) } if (magicmask1 & (1 << 2)) == 0 { diff --git a/libraries/corereader/slice_reader.go b/libraries/corereader/slice_reader.go index 85ddde8..78e44fc 100644 --- a/libraries/corereader/slice_reader.go +++ b/libraries/corereader/slice_reader.go @@ -137,7 +137,6 @@ func (sm *SharedSliceMetadata) markSliceFinalized(idx int) { } } - // findSliceForGlob finds the slice containing the given global sequence using binary search. // Returns the slice info and index, or nil/-1 if not found. func (sm *SharedSliceMetadata) findSliceForGlob(glob uint64) (*SliceInfo, int) { @@ -366,17 +365,17 @@ type SliceInfo struct { } type sliceReader struct { - sliceNum uint32 - basePath string - blockIndex map[uint32]blockIndexEntry - minBlock uint32 // Cached minimum block number from blockIndex - blockFile *os.File - mmapData []byte - header *DataLogHeader - loaded bool - mu sync.RWMutex - refCount atomic.Int32 - mmapWaitTimeout time.Duration // Max time to wait for data file growth (0 = use default 30s) + sliceNum uint32 + basePath string + blockIndex map[uint32]blockIndexEntry + minBlock uint32 // Cached minimum block number from blockIndex + blockFile *os.File + mmapData []byte + header *DataLogHeader + loaded bool + mu sync.RWMutex + refCount atomic.Int32 + mmapWaitTimeout time.Duration // Max time to wait for data file growth (0 = use default 30s) } type SliceGlobRange struct { @@ -3650,6 +3649,7 @@ func parseBlockWithCanonical(blockData []byte, filterFunc ActionFilterFunc) (map action := CanonicalAction{ ActionOrdinal: act.ActionOrdinal, CreatorAO: act.CreatorAO, + ClosestUAAO: act.ClosestUAAO, ReceiverUint64: blob.Block.NamesInBlock[act.ReceiverIndex], DataIndex: act.DataIndex, AuthAccountIndexes: authSlice, diff --git a/libraries/corereader/types.go b/libraries/corereader/types.go index ae98e70..d2b749e 100644 --- a/libraries/corereader/types.go +++ b/libraries/corereader/types.go @@ -16,6 +16,7 @@ type ActionMetadata struct { type CanonicalAction struct { ActionOrdinal uint32 CreatorAO uint32 + ClosestUAAO uint32 ReceiverUint64 uint64 DataIndex uint32 AuthAccountIndexes []uint32 @@ -66,13 +67,16 @@ func (r *RawBlock) SetActionData(rawData []byte, offsets, lengths []uint32) { // 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 { - Account uint64 - Contract uint64 - Action uint64 - GlobalSeq uint64 - TrxIndex uint32 - IsAuthorizer bool - Receiver uint64 // chain receiver; differs from Account on authorizer-indexed entries + Account uint64 + Contract uint64 + Action uint64 + GlobalSeq uint64 + TrxIndex uint32 + IsAuthorizer bool + Receiver uint64 // chain receiver; differs from Account on authorizer-indexed entries + ActionOrdinal uint32 + CreatorAO uint32 + ClosestUAAO uint32 } // ContractExecution represents a contract execution (receiver == contract). diff --git a/services/actionindex/internal/broadcaster.go b/services/actionindex/internal/broadcaster.go index c081d42..c6c05e3 100644 --- a/services/actionindex/internal/broadcaster.go +++ b/services/actionindex/internal/broadcaster.go @@ -9,15 +9,18 @@ import ( ) type StreamedAction struct { - GlobalSeq uint64 - BlockNum uint32 - BlockTime uint32 - Contract uint64 - Action uint64 - Receiver uint64 - ActionData []byte - CpuUsageUs uint32 - NetUsageWords uint32 + GlobalSeq uint64 + BlockNum uint32 + BlockTime uint32 + Contract uint64 + Action uint64 + Receiver uint64 + ActionData []byte + CpuUsageUs uint32 + NetUsageWords uint32 + ActionOrdinal uint32 + CreatorActionOrdinal uint32 + ClosestUnnotifiedAncestorActionOrdinal uint32 } type ActionBroadcaster struct { diff --git a/services/actionindex/internal/stream_catchup.go b/services/actionindex/internal/stream_catchup.go index cf4ae3d..bfdf3f3 100644 --- a/services/actionindex/internal/stream_catchup.go +++ b/services/actionindex/internal/stream_catchup.go @@ -241,14 +241,17 @@ func (c *StreamCatchup) processBatch(seqs []uint64, account uint64, seqToAccount matchedViaBuf[0] = 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), - CpuUsageUs: at.CpuUsageUs, - NetUsageWords: at.NetUsageWords, + 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, + ActionOrdinal: at.ActionOrdinal, + CreatorActionOrdinal: at.CreatorAO, + ClosestUnnotifiedAncestorActionOrdinal: at.ClosestUAAO, } if !c.filter.Matches(action, matchedViaBuf[:]) { diff --git a/services/actionindex/internal/stream_catchup_test.go b/services/actionindex/internal/stream_catchup_test.go index c32b9e0..dfd3eb3 100644 --- a/services/actionindex/internal/stream_catchup_test.go +++ b/services/actionindex/internal/stream_catchup_test.go @@ -9,8 +9,8 @@ import ( type stubBaseReader struct { corereader.BaseReader - traces []chain.ActionTrace - bySeq map[uint64]chain.ActionTrace + traces []chain.ActionTrace + bySeq map[uint64]chain.ActionTrace } func (s *stubBaseReader) GetActionsByGlobalSeqs(seqs []uint64) ([]chain.ActionTrace, *corereader.FetchTimings, error) { @@ -114,6 +114,45 @@ func TestProcessBatch_DeliversReceiverPathAction(t *testing.T) { } } +func TestProcessBatch_CarriesOrdinals(t *testing.T) { + tracked := chain.StringToName("shipload.gm") + + trace := chain.ActionTrace{ + ActionOrdinal: 3, + CreatorAO: 1, + ClosestUAAO: 1, + Receiver: "shipload.gm", + Act: chain.Action{ + Account: "shipload.gm", + Name: "consume", + }, + } + + catchup := &StreamCatchup{ + reader: &stubBaseReader{traces: []chain.ActionTrace{trace}}, + filter: ActionFilter{ + Receivers: map[uint64]struct{}{tracked: {}}, + }, + } + + var delivered []StreamedAction + sent, err := catchup.processBatch([]uint64{900}, 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.ActionOrdinal != 3 || got.CreatorActionOrdinal != 1 || got.ClosestUnnotifiedAncestorActionOrdinal != 1 { + t.Errorf("ordinals = (%d,%d,%d), want (3,1,1)", + got.ActionOrdinal, got.CreatorActionOrdinal, got.ClosestUnnotifiedAncestorActionOrdinal) + } +} + func TestProcessBatch_MultiAccount_AttributesPerSeq(t *testing.T) { shipload := chain.StringToName("shipload.gm") platform := chain.StringToName("platform.gm") diff --git a/services/actionindex/internal/stream_tcp.go b/services/actionindex/internal/stream_tcp.go index fa53f30..78a6299 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, 48+len(actionData)) + payload := make([]byte, 60+len(actionData)) binary.LittleEndian.PutUint64(payload[0:8], action.GlobalSeq) binary.LittleEndian.PutUint32(payload[8:12], action.BlockNum) @@ -308,7 +308,10 @@ func (ts *StreamTCPServer) sendAction(conn net.Conn, action StreamedAction, deco binary.LittleEndian.PutUint64(payload[32:40], action.Receiver) binary.LittleEndian.PutUint32(payload[40:44], action.CpuUsageUs) binary.LittleEndian.PutUint32(payload[44:48], action.NetUsageWords) - copy(payload[48:], actionData) + binary.LittleEndian.PutUint32(payload[48:52], action.ActionOrdinal) + binary.LittleEndian.PutUint32(payload[52:56], action.CreatorActionOrdinal) + binary.LittleEndian.PutUint32(payload[56:60], action.ClosestUnnotifiedAncestorActionOrdinal) + copy(payload[60:], actionData) return ts.writeMessage(conn, msgType, payload) } diff --git a/services/actionindex/internal/stream_tcp_test.go b/services/actionindex/internal/stream_tcp_test.go new file mode 100644 index 0000000..dab4646 --- /dev/null +++ b/services/actionindex/internal/stream_tcp_test.go @@ -0,0 +1,60 @@ +package internal + +import ( + "encoding/binary" + "io" + "net" + "testing" +) + +func TestSendActionLayout(t *testing.T) { + serverConn, clientConn := net.Pipe() + defer clientConn.Close() + + ts := &StreamTCPServer{} + action := StreamedAction{ + GlobalSeq: 42, + BlockNum: 7, + BlockTime: 1700000000, + Contract: 0x1111, + Action: 0x2222, + Receiver: 0x3333, + CpuUsageUs: 11, + NetUsageWords: 22, + ActionOrdinal: 3, + CreatorActionOrdinal: 1, + ClosestUnnotifiedAncestorActionOrdinal: 1, + ActionData: []byte("abc"), + } + + go func() { + _ = ts.sendAction(serverConn, action, false) + serverConn.Close() + }() + + header := make([]byte, 5) + if _, err := io.ReadFull(clientConn, header); err != nil { + t.Fatalf("read header: %v", err) + } + length := binary.BigEndian.Uint32(header[0:4]) + payload := make([]byte, length-1) + if _, err := io.ReadFull(clientConn, payload); err != nil { + t.Fatalf("read payload: %v", err) + } + + if len(payload) != 60+3 { + t.Fatalf("payload len = %d, want 63", len(payload)) + } + if got := binary.LittleEndian.Uint32(payload[48:52]); got != 3 { + t.Errorf("ActionOrdinal = %d, want 3", got) + } + if got := binary.LittleEndian.Uint32(payload[52:56]); got != 1 { + t.Errorf("CreatorActionOrdinal = %d, want 1", got) + } + if got := binary.LittleEndian.Uint32(payload[56:60]); got != 1 { + t.Errorf("ClosestUnnotifiedAncestorActionOrdinal = %d, want 1", got) + } + if string(payload[60:]) != "abc" { + t.Errorf("action_data = %q, want \"abc\"", payload[60:]) + } +} diff --git a/services/actionindex/internal/sync.go b/services/actionindex/internal/sync.go index d2743c4..bc28540 100644 --- a/services/actionindex/internal/sync.go +++ b/services/actionindex/internal/sync.go @@ -348,15 +348,18 @@ func (p *AccountHistoryProcessor) broadcastActions(block corereader.Block) error 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, + GlobalSeq: seq, + BlockNum: block.BlockNum, + BlockTime: block.BlockTime, + Contract: a.Contract, + Action: a.Action, + Receiver: a.Receiver, + ActionData: data, + CpuUsageUs: cpuUs, + NetUsageWords: netWords, + ActionOrdinal: a.ActionOrdinal, + CreatorActionOrdinal: a.CreatorAO, + ClosestUnnotifiedAncestorActionOrdinal: a.ClosestUAAO, }, g.matchedVia) } diff --git a/services/actionindex/internal/sync_broadcast_test.go b/services/actionindex/internal/sync_broadcast_test.go index 5b8948b..115ab57 100644 --- a/services/actionindex/internal/sync_broadcast_test.go +++ b/services/actionindex/internal/sync_broadcast_test.go @@ -8,6 +8,56 @@ import ( "github.com/greymass/roborovski/libraries/corereader" ) +func TestBroadcastActions_CarriesOrdinals(t *testing.T) { + contract := chain.StringToName("shipload.gm") + action := chain.StringToName("consume") + const seq = uint64(700) + + broadcaster := NewActionBroadcaster() + broadcaster.SetLiveMode(true) + sub := broadcaster.Subscribe(ActionFilter{ + Contracts: map[uint64]struct{}{contract: {}}, + }) + + reader := &stubReaderForBroadcast{ + bySeq: map[uint64]chain.ActionTrace{ + seq: { + BlockNum: 1, BlockTime: "1970-01-01T00:16:40.000", + Receiver: "shipload.gm", + Act: chain.Action{Account: "shipload.gm", Name: "consume"}, + }, + }, + } + + block := corereader.Block{ + BlockNum: 1, + BlockTime: 1000, + MaxSeq: seq, + Actions: []corereader.Action{ + { + Account: contract, Contract: contract, Action: action, + GlobalSeq: seq, Receiver: contract, + ActionOrdinal: 3, CreatorAO: 1, ClosestUAAO: 1, + }, + }, + } + + 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)) + } + d := delivered[0] + if d.ActionOrdinal != 3 || d.CreatorActionOrdinal != 1 || d.ClosestUnnotifiedAncestorActionOrdinal != 1 { + t.Errorf("ordinals = (%d,%d,%d), want (3,1,1)", + d.ActionOrdinal, d.CreatorActionOrdinal, d.ClosestUnnotifiedAncestorActionOrdinal) + } +} // stubReaderForBroadcast satisfies corereader.Reader minimally for broadcastActions tests. // Only GetActionsByGlobalSeqs is called from the fetched-traces sub-path.