Add slot indexing

car-dag-traverser.go

@@ -251,12 +251,20 @@ func (t *SimpleIterator) GetTransaction(ctx context.Context, c cid.Cid) (*ipldbi
 // It stops iterating if the callback returns an error.
 // It works by iterating over all objects in the CAR file and
 // calling the callback for each object that is a Subset.
-func (t *SimpleIterator) FindSubsets(ctx context.Context, callback func(*ipldbindcode.Subset) error) error {
+func (t *SimpleIterator) FindSubsets(ctx context.Context, callback func(cid.Cid, *ipldbindcode.Subset) error) error {
 	dr, err := t.cr.DataReader()
 	if err != nil {
 		return fmt.Errorf("failed to get data reader: %w", err)
 	}
-	rd, err := car.NewCarReader(dr)
+	return FindSubsets(ctx, dr, callback)
+}
+
+func FindSubsets(
+	ctx context.Context,
+	sectionReader carv2.SectionReader,
+	callback func(cid.Cid, *ipldbindcode.Subset) error,
+) error {
+	rd, err := car.NewCarReader(sectionReader)
 	if err != nil {
 		return fmt.Errorf("failed to create car reader: %w", err)
 	}
@@ -274,7 +282,7 @@ func (t *SimpleIterator) FindSubsets(ctx context.Context, callback func(*ipldbin
 				// TODO: log error, or return error?
 				continue
 			}
-			err = callback(decoded)
+			err = callback(block.Cid(), decoded)
 			if err != nil {
 				if err == ErrStopIteration {
 					return nil
@@ -292,12 +300,20 @@ var ErrStopIteration = errors.New("stop iteration")
 // It stops iterating if the callback returns an error.
 // It works by iterating over all objects in the CAR file and
 // calling the callback for each object that is a Block.
-func (t *SimpleIterator) FindBlocks(ctx context.Context, callback func(*ipldbindcode.Block) error) error {
+func (t *SimpleIterator) FindBlocks(ctx context.Context, callback func(cid.Cid, *ipldbindcode.Block) error) error {
 	dr, err := t.cr.DataReader()
 	if err != nil {
 		return fmt.Errorf("failed to get data reader: %w", err)
 	}
-	rd, err := car.NewCarReader(dr)
+	return FindBlocks(ctx, dr, callback)
+}
+
+func FindBlocks(
+	ctx context.Context,
+	sectionReader carv2.SectionReader,
+	callback func(cid.Cid, *ipldbindcode.Block) error,
+) error {
+	rd, err := car.NewCarReader(sectionReader)
 	if err != nil {
 		return fmt.Errorf("failed to create car reader: %w", err)
 	}
@@ -314,7 +330,7 @@ func (t *SimpleIterator) FindBlocks(ctx context.Context, callback func(*ipldbind
 			if err != nil {
 				continue
 			}
-			err = callback(decoded)
+			err = callback(block.Cid(), decoded)
 			if err != nil {
 				if err == ErrStopIteration {
 					return nil
@@ -330,12 +346,20 @@ func (t *SimpleIterator) FindBlocks(ctx context.Context, callback func(*ipldbind
 // It stops iterating if the callback returns an error.
 // It works by iterating over all objects in the CAR file and
 // calling the callback for each object that is an Entry.
-func (t *SimpleIterator) FindEntries(ctx context.Context, callback func(*ipldbindcode.Entry) error) error {
+func (t *SimpleIterator) FindEntries(ctx context.Context, callback func(cid.Cid, *ipldbindcode.Entry) error) error {
 	dr, err := t.cr.DataReader()
 	if err != nil {
 		return fmt.Errorf("failed to get data reader: %w", err)
 	}
-	rd, err := car.NewCarReader(dr)
+	return FindEntries(ctx, dr, callback)
+}
+
+func FindEntries(
+	ctx context.Context,
+	sectionReader carv2.SectionReader,
+	callback func(cid.Cid, *ipldbindcode.Entry) error,
+) error {
+	rd, err := car.NewCarReader(sectionReader)
 	if err != nil {
 		return fmt.Errorf("failed to create car reader: %w", err)
 	}
@@ -352,7 +376,7 @@ func (t *SimpleIterator) FindEntries(ctx context.Context, callback func(*ipldbin
 			if err != nil {
 				continue
 			}
-			err = callback(decoded)
+			err = callback(block.Cid(), decoded)
 			if err != nil {
 				return err
 			}
@@ -365,12 +389,20 @@ func (t *SimpleIterator) FindEntries(ctx context.Context, callback func(*ipldbin
 // It stops iterating if the callback returns an error.
 // It works by iterating over all objects in the CAR file and
 // calling the callback for each object that is a Transaction.
-func (t *SimpleIterator) FindTransactions(ctx context.Context, callback func(*ipldbindcode.Transaction) error) error {
+func (t *SimpleIterator) FindTransactions(ctx context.Context, callback func(cid.Cid, *ipldbindcode.Transaction) error) error {
 	dr, err := t.cr.DataReader()
 	if err != nil {
 		return fmt.Errorf("failed to get data reader: %w", err)
 	}
-	rd, err := car.NewCarReader(dr)
+	return FindTransactions(ctx, dr, callback)
+}
+
+func FindTransactions(
+	ctx context.Context,
+	sectionReader carv2.SectionReader,
+	callback func(cid.Cid, *ipldbindcode.Transaction) error,
+) error {
+	rd, err := car.NewCarReader(sectionReader)
 	if err != nil {
 		return fmt.Errorf("failed to create car reader: %w", err)
 	}
@@ -387,7 +419,7 @@ func (t *SimpleIterator) FindTransactions(ctx context.Context, callback func(*ip
 			if err != nil {
 				continue
 			}
-			err = callback(decoded)
+			err = callback(block.Cid(), decoded)
 			if err != nil {
 				return err
 			}
@@ -396,6 +428,37 @@ func (t *SimpleIterator) FindTransactions(ctx context.Context, callback func(*ip
 	return nil
 }
 
+func FindAny(
+	ctx context.Context,
+	sectionReader carv2.SectionReader,
+	callback func(cid.Cid, any) error,
+) error {
+	rd, err := car.NewCarReader(sectionReader)
+	if err != nil {
+		return fmt.Errorf("failed to create car reader: %w", err)
+	}
+	for {
+		block, err := rd.Next()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		{
+			decoded, err := iplddecoders.DecodeAny(block.RawData())
+			if err != nil {
+				continue
+			}
+			err = callback(block.Cid(), decoded)
+			if err != nil {
+				if err == ErrStopIteration {
+					return nil
+				}
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 type offsetFinderFunc func(ctx context.Context, c cid.Cid) (uint64, error)
 
 func getRawNodeFromCarByCid(offsetFinder offsetFinderFunc, cr *carv2.Reader, c cid.Cid) (*blocks.BasicBlock, error) {

cmd-x-index-cid2offset.go

@@ -8,7 +8,7 @@ import (
 	"k8s.io/klog/v2"
 )
 
-func newCmd_Index_Cid2Offset() *cli.Command {
+func newCmd_Index_cid2offset() *cli.Command {
 	var verify bool
 	return &cli.Command{
 		Name:        "cid-to-offset",
@@ -34,7 +34,7 @@ func newCmd_Index_Cid2Offset() *cli.Command {
 				defer func() {
 					klog.Infof("Finished in %s", time.Since(startedAt))
 				}()
-				klog.Infof("Creating index for %s", carPath)
+				klog.Infof("Creating CID-to-offset index for %s", carPath)
 				indexFilepath, err := CreateIndex_cid2offset(
 					context.TODO(),
 					carPath,

cmd-x-index-slot2cid.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/urfave/cli/v2"
+	"k8s.io/klog/v2"
+)
+
+func newCmd_Index_slot2cid() *cli.Command {
+	var verify bool
+	return &cli.Command{
+		Name:        "slot-to-cid",
+		Description: "Given a CAR file containing a Solana epoch, create an index of the file that maps slot numbers to CIDs.",
+		ArgsUsage:   "<car-path> <index-dir>",
+		Before: func(c *cli.Context) error {
+			return nil
+		},
+		Flags: []cli.Flag{
+			&cli.BoolFlag{
+				Name:        "verify",
+				Usage:       "verify the index after creating it",
+				Destination: &verify,
+			},
+		},
+		Subcommands: []*cli.Command{},
+		Action: func(c *cli.Context) error {
+			carPath := c.Args().Get(0)
+			indexDir := c.Args().Get(1)
+
+			{
+				startedAt := time.Now()
+				defer func() {
+					klog.Infof("Finished in %s", time.Since(startedAt))
+				}()
+				klog.Infof("Creating Slot-to-CID index for %s", carPath)
+				indexFilepath, err := CreateIndex_slot2cid(
+					context.TODO(),
+					carPath,
+					indexDir,
+				)
+				if err != nil {
+					panic(err)
+				}
+				klog.Info("Index created")
+				if verify {
+					klog.Infof("Verifying index for %s located at %s", carPath, indexFilepath)
+					startedAt := time.Now()
+					defer func() {
+						klog.Infof("Finished in %s", time.Since(startedAt))
+					}()
+					err := VerifyIndex_slot2cid(context.TODO(), carPath, indexFilepath)
+					if err != nil {
+						return cli.Exit(err, 1)
+					}
+					klog.Info("Index verified")
+					return nil
+				}
+			}
+			return nil
+		},
+	}
+}

cmd-x-index.go

@@ -13,7 +13,8 @@ func newCmd_Index() *cli.Command {
 		},
 		Flags: []cli.Flag{},
 		Subcommands: []*cli.Command{
-			newCmd_Index_Cid2Offset(),
+			newCmd_Index_cid2offset(),
+			newCmd_Index_slot2cid(),
 		},
 	}
 }

cmd-x-traverse.go

@@ -8,6 +8,7 @@ import (
 	"github.com/davecgh/go-spew/spew"
 	bin "github.com/gagliardetto/binary"
 	"github.com/gagliardetto/solana-go"
+	"github.com/ipfs/go-cid"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	"github.com/urfave/cli/v2"
 	"go.firedancer.io/radiance/cmd/radiance/car/createcar/ipld/ipldbindcode"
@@ -123,7 +124,7 @@ func newCmd_XTraverse() *cli.Command {
 
 				if false {
 					klog.Info("Iterating Solana blocks...")
-					err = simpleIter.FindBlocks(context.Background(), func(block *ipldbindcode.Block) error {
+					err = simpleIter.FindBlocks(context.Background(), func(_ cid.Cid, block *ipldbindcode.Block) error {
 						numSolanaBlocks++
 						if numSolanaBlocks%10_000 == 0 {
 							fmt.Print(".")
@@ -137,7 +138,7 @@ func newCmd_XTraverse() *cli.Command {
 					klog.Infof("Finished iterating blocks in %s; found %d solana blocks", took, numSolanaBlocks)
 
 					klog.Info("Iterating Solana Transactions...")
-					err = simpleIter.FindTransactions(context.Background(), func(tx *ipldbindcode.Transaction) error {
+					err = simpleIter.FindTransactions(context.Background(), func(_ cid.Cid, tx *ipldbindcode.Transaction) error {
 						numTransactions++
 						if numTransactions%100_000 == 0 {
 							fmt.Print(".")

cmd-x-verify-index-cid2offset.go

@@ -8,7 +8,7 @@ import (
 	"k8s.io/klog/v2"
 )
 
-func newCmd_VerifyIndex_Cid2Offset() *cli.Command {
+func newCmd_VerifyIndex_cid2offset() *cli.Command {
 	return &cli.Command{
 		Name:        "cid-to-offset",
 		Description: "Verify the index of the CAR file that maps CIDs to offsets in the CAR file.",
@@ -25,7 +25,7 @@ func newCmd_VerifyIndex_Cid2Offset() *cli.Command {
 				defer func() {
 					klog.Infof("Finished in %s", time.Since(startedAt))
 				}()
-				klog.Infof("Verifying index for %s", carPath)
+				klog.Infof("Verifying CID-to-offset index for %s", carPath)
 				err := VerifyIndex_cid2offset(context.TODO(), carPath, indexFilePath)
 				if err != nil {
 					return err

cmd-x-verify-index-slot2cid.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/urfave/cli/v2"
+	"k8s.io/klog/v2"
+)
+
+func newCmd_VerifyIndex_slot2cid() *cli.Command {
+	return &cli.Command{
+		Name:        "slot-to-cid",
+		Description: "Verify the index of the CAR file that maps slot numbers to CIDs.",
+		ArgsUsage:   "<car-path> <index-file-path>",
+		Before: func(c *cli.Context) error {
+			return nil
+		},
+		Flags: []cli.Flag{},
+		Action: func(c *cli.Context) error {
+			carPath := c.Args().Get(0)
+			indexFilePath := c.Args().Get(1)
+			{
+				startedAt := time.Now()
+				defer func() {
+					klog.Infof("Finished in %s", time.Since(startedAt))
+				}()
+				klog.Infof("Verifying Slot-to-CID index for %s", carPath)
+				err := VerifyIndex_slot2cid(context.TODO(), carPath, indexFilePath)
+				if err != nil {
+					return err
+				}
+				klog.Info("Index verified successfully")
+			}
+			return nil
+		},
+	}
+}

cmd-x-verify-index.go

@@ -13,7 +13,8 @@ func newCmd_VerifyIndex() *cli.Command {
 		},
 		Flags: []cli.Flag{},
 		Subcommands: []*cli.Command{
-			newCmd_VerifyIndex_Cid2Offset(),
+			newCmd_VerifyIndex_cid2offset(),
+			newCmd_VerifyIndex_slot2cid(),
 		},
 	}
 }

compactindex36/compactindex.go

@@ -239,7 +239,7 @@ type Entry struct {
 
 // intWidth returns the number of bytes minimally required to represent the given integer.
 func intWidth(n uint64) uint8 {
-	return 36
+	return 36 // 36 is the length of the CIDs we use.
 	msb := 64 - bits.LeadingZeros64(n)
 	return uint8((msb + 7) / 8)
 }