Fix epoch calc for range (#296)

* Fix epoch calc for range

* Move certain log messages to v 4

* Remove redundant wait

* gsfa: check index size

* Cleanup

Author: gagliardetto

grpc-server.go

@@ -187,9 +187,9 @@ func (multi *MultiEpoch) GetBlock(ctx context.Context, params *old_faithful_grpc
 				return err
 			}
 			if slot == 0 {
-				klog.V(5).Infof("car start to slot(0)::%s", blockCid)
+				klog.V(4).Infof("car start to slot(0)::%s", blockCid)
 			} else {
-				klog.V(5).Infof(
+				klog.V(4).Infof(
 					"slot(%d)::%s to slot(%d)::%s",
 					uint64(block.Meta.Parent_slot),
 					parentBlockCid,
@@ -262,7 +262,7 @@ func (multi *MultiEpoch) GetBlock(ctx context.Context, params *old_faithful_grpc
 					attribute.Int64("read_length", int64(length)),
 				)
 
-				klog.V(5).Infof("prefetching CAR: start=%d length=%d (parent_offset=%d)", start, length, parentOffset)
+				klog.V(4).Infof("prefetching CAR: start=%d length=%d (parent_offset=%d)", start, length, parentOffset)
 
 				// This is the actual disk read operation - likely significant seek time here
 				readCtx, readSpan := telemetry.StartDiskIOSpan(prefetchCtx, "read_car_section", map[string]string{
@@ -548,7 +548,7 @@ func (multi *MultiEpoch) GetBlock(ctx context.Context, params *old_faithful_grpc
 			}
 		} else {
 			if slot != 0 {
-				klog.V(5).Infof("parent slot is in a different epoch, not implemented yet (can't get previousBlockhash)")
+				klog.V(4).Infof("parent slot is in a different epoch, not implemented yet (can't get previousBlockhash)")
 			}
 		}
 		parentSpan.End()
@@ -899,13 +899,18 @@ func (multi *MultiEpoch) StreamTransactions(params *old_faithful_grpc.StreamTran
 		endSlot = *params.EndSlot
 	}
 	gsfaReader, epochNums := multi.getGsfaReadersInEpochDescendingOrderForSlotRange(ctx, startSlot, endSlot)
+	wantedEpochs := slottools.CalcEpochsForSlotRange(startSlot, endSlot)
+	klog.V(4).Infof("Streaming transactions from slots %d to %d, epochs %v", startSlot, endSlot, wantedEpochs)
 
 	gsfaReadersLoaded := true
 	if len(epochNums) == 0 {
-		klog.V(5).Info("The requested slot range does not have any GSFA readers loaded, will use the default method")
+		klog.V(4).Info("The requested slot range does not have any GSFA readers loaded, will use the default method")
 		gsfaReadersLoaded = false
 	} else {
-		klog.V(5).Infof("Using GSFA readers for epochs: %v", epochNums)
+		klog.V(4).Infof("Using GSFA readers for epochs: %v; wanted epochs: %v", epochNums, wantedEpochs)
+		if len(epochNums) < len(wantedEpochs) {
+			klog.V(4).Infof("Not all epochs in the requested slot range have GSFA readers loaded")
+		}
 	}
 
 	return multi.processSlotTransactions(ctx, ser, startSlot, endSlot, params.Filter, gsfaReader, gsfaReadersLoaded)
@@ -930,7 +935,7 @@ func (multi *MultiEpoch) processSlotTransactions(
 	}
 
 	if filter == nil || len(filter.AccountInclude) == 0 || !gsfaReadersLoaded {
-		klog.V(5).Infof("Using the old faithful method for streaming transactions from slots %d to %d", startSlot, endSlot)
+		klog.V(4).Infof("Using the old faithful method for streaming transactions from slots %d to %d", startSlot, endSlot)
 
 		for slot := startSlot; slot <= endSlot; slot++ {
 			select {
@@ -999,7 +1004,7 @@ func (multi *MultiEpoch) processSlotTransactions(
 		return nil
 	}
 
-	klog.V(5).Infof("Using GSFA reader for streaming transactions from slots %d to %d", startSlot, endSlot)
+	klog.V(4).Infof("Using GSFA reader for streaming transactions from slots %d to %d", startSlot, endSlot)
 
 	const batchSize = 100
 	buffer := newTxBuffer(uint64(startSlot), uint64(endSlot))
@@ -1053,7 +1058,7 @@ func (multi *MultiEpoch) processSlotTransactions(
 					return err
 				}
 				duration := time.Since(startTime)
-				klog.V(5).Infof("GSFA query completed for account %s, from slot %d to %d took %s", pKey.String(), startSlot, endSlot, duration)
+				klog.V(4).Infof("GSFA query completed for account %s, from slot %d to %d took %s", pKey.String(), startSlot, endSlot, duration)
 
 				for epochNumber, txns := range epochToTxns {
 					epochHandler, err := multi.GetEpoch(epochNumber)
@@ -1118,7 +1123,6 @@ func (multi *MultiEpoch) processSlotTransactions(
 	// Handle any errors
 	klog.V(5).Infof("Checking for errors from goroutines")
 	errCheckStartTime := time.Now()
-	wg.Wait()
 	select {
 	case err := <-wgWaitToChannel(wg):
 		if err != nil {
@@ -1139,11 +1143,11 @@ func (multi *MultiEpoch) processSlotTransactions(
 	if err := buffer.flush(ser); err != nil {
 		return err
 	}
-	klog.V(5).Infof("Buffer flush completed in %s", time.Since(flushStartTime))
+	klog.V(4).Infof("Buffer flush completed in %s", time.Since(flushStartTime))
 
 	// If we got here with no transactions (buffer is empty), send an empty response
 	if len(buffer.items) == 0 {
-		klog.V(5).Infof("No transactions found for the requested accounts, sending empty response")
+		klog.V(4).Infof("No transactions found for the requested accounts, sending empty response")
 		emptyResp := &old_faithful_grpc.TransactionResponse{
 			Slot: startSlot,
 			// Include other required fields as needed

gsfa/gsfa-read.go

@@ -30,6 +30,17 @@ func isDir(path string) (bool, error) {
 	return info.IsDir(), nil
 }
 
+func sizeOfFile(path string) (int64, error) {
+	info, err := os.Stat(path)
+	if err != nil {
+		return 0, fmt.Errorf("error while getting file info for %s: %w", path, err)
+	}
+	if info.IsDir() {
+		return 0, fmt.Errorf("expected a file, but got a directory: %s", path)
+	}
+	return info.Size(), nil
+}
+
 // NewGsfaReader opens an existing index in READ-ONLY mode.
 func NewGsfaReader(indexRootDir string) (*GsfaReader, error) {
 	if ok, err := isDir(indexRootDir); err != nil {
@@ -40,6 +51,19 @@ func NewGsfaReader(indexRootDir string) (*GsfaReader, error) {
 	index := &GsfaReader{}
 	{
 		offsetsIndex := filepath.Join(indexRootDir, string(indexes.Kind_PubkeyToOffsetAndSize)+".index")
+		// check size; if is 1733 bytes, then it's the buggy version of the index.
+		size, err := sizeOfFile(offsetsIndex)
+		if err != nil {
+			return nil, fmt.Errorf("error while getting size of offsets index file %s: %w", offsetsIndex, err)
+		}
+		if size == 1733 {
+			return nil, fmt.Errorf("the offsets index file %s is the buggy version (1733 bytes); "+
+				"please delete the whole gsfa index and re-create it with the latest old-faithful-cli", offsetsIndex)
+		} else if size < 1733 {
+			return nil, fmt.Errorf("the offsets index file %s is too small (%d bytes); "+
+				"please delete the whole gsfa index and re-create it with the latest old-faithful-cli. "+
+				"Expected at least 1733 bytes, got %d bytes", offsetsIndex, size, size)
+		}
 		offsets, err := indexes.Open_PubkeyToOffsetAndSize(offsetsIndex)
 		if err != nil {
 			return nil, fmt.Errorf("error while opening offsets index: %w", err)

multiepoch-getSignaturesForAddress.go

@@ -48,13 +48,13 @@ func (ser *MultiEpoch) getGsfaReadersInEpochDescendingOrderForSlotRange(ctx cont
 	ser.mu.RLock()
 	defer ser.mu.RUnlock()
 
-	startEpoch := slottools.CalcEpochForSlot(startSlot)
-	endEpoch := slottools.CalcEpochForSlot(endSlot)
-
-	epochs := make([]*Epoch, 0, endEpoch-startEpoch+1)
-	for _, epoch := range ser.epochs {
-		if epoch.Epoch() >= startEpoch && epoch.Epoch() <= endEpoch {
+	wantedEpochs := slottools.CalcEpochsForSlotRange(startSlot, endSlot)
+	epochs := make([]*Epoch, 0, len(wantedEpochs))
+	for _, wantedEpoch := range wantedEpochs {
+		if epoch, ok := ser.epochs[wantedEpoch]; ok {
 			epochs = append(epochs, epoch)
+		} else {
+			klog.Warningf("epoch %d not found in multiepoch", wantedEpoch)
 		}
 	}
 

slottools/edges.go

@@ -44,3 +44,23 @@ func Uint64ToLEBytes(v uint64) []byte {
 func Uint64FromLEBytes(buf []byte) uint64 {
 	return binary.LittleEndian.Uint64(buf)
 }
+
+func CalcEpochsForSlotRange(startSlot, endSlot uint64) []uint64 {
+	epochStart := CalcEpochForSlot(startSlot)
+	epochEnd := CalcEpochForSlot(endSlot)
+	return calcRangeInclusive(epochStart, epochEnd)
+}
+
+func calcRangeInclusive(start, end uint64) []uint64 {
+	if start == end {
+		return []uint64{start} // if start and end are the same, return a slice with that single value
+	}
+	if start > end {
+		end, start = start, end // ensure start is less than or equal to end
+	}
+	rangeSlice := make([]uint64, end-start+1)
+	for i := range rangeSlice {
+		rangeSlice[i] = start + uint64(i)
+	}
+	return rangeSlice
+}

slottools/edges_test.go

@@ -1,6 +1,8 @@
 package slottools
 
 import (
+	"fmt"
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -66,3 +68,54 @@ func TestUint64RangesHavePartialOverlapIncludingEdges(t *testing.T) {
 		require.True(t, Uint64RangesHavePartialOverlapIncludingEdges(r1, r2))
 	}
 }
+
+func TestRanges(t *testing.T) {
+	got := calcRangeInclusive(0, 5)
+	if !reflect.DeepEqual(got, []uint64{0, 1, 2, 3, 4, 5}) {
+		panic(fmt.Sprintf("calcRangeInclusive(0, 5) = %v, want [0 1 2 3 4 5]", got))
+	}
+	got = calcRangeInclusive(5, 0)
+	if !reflect.DeepEqual(got, []uint64{0, 1, 2, 3, 4, 5}) {
+		panic(fmt.Sprintf("calcRangeInclusive(5, 0) = %v, want [0 1 2 3 4 5]", got))
+	}
+	got = calcRangeInclusive(3, 3)
+	if !reflect.DeepEqual(got, []uint64{3}) {
+		panic(fmt.Sprintf("calcRangeInclusive(3, 3) = %v, want [3]", got))
+	}
+	got = calcRangeInclusive(0, 0)
+	if !reflect.DeepEqual(got, []uint64{0}) {
+		panic(fmt.Sprintf("calcRangeInclusive(0, 0) = %v, want [0]", got))
+	}
+	got = calcRangeInclusive(10, 15)
+	if !reflect.DeepEqual(got, []uint64{10, 11, 12, 13, 14, 15}) {
+		panic(fmt.Sprintf("calcRangeInclusive(10, 15) = %v, want [10 11 12 13 14 15]", got))
+	}
+	got = calcRangeInclusive(15, 10)
+	if !reflect.DeepEqual(got, []uint64{10, 11, 12, 13, 14, 15}) {
+		panic(fmt.Sprintf("calcRangeInclusive(15, 10) = %v, want [10 11 12 13 14 15]", got))
+	}
+}
+
+func TestEpochRange(t *testing.T) {
+	{
+		gotEpochs := CalcEpochsForSlotRange(0, 431999)
+		wantEpochs := []uint64{0}
+		if !reflect.DeepEqual(gotEpochs, wantEpochs) {
+			t.Errorf("CalcEpochsForSlotRange(0, 431999) = %v, want %v", gotEpochs, wantEpochs)
+		}
+	}
+	{
+		gotEpochs := CalcEpochsForSlotRange(432000, 863999)
+		wantEpochs := []uint64{1}
+		if !reflect.DeepEqual(gotEpochs, wantEpochs) {
+			t.Errorf("CalcEpochsForSlotRange(432000, 863999) = %v, want %v", gotEpochs, wantEpochs)
+		}
+	}
+	{
+		gotEpochs := CalcEpochsForSlotRange(343169187, 345302213)
+		wantEpochs := []uint64{
+			794, 795, 796, 797, 798, 799,
+		}
+		require.Equal(t, wantEpochs, gotEpochs, "CalcEpochsForSlotRange(343169187, 345302213) should return the expected epochs")
+	}
+}