blockstore: replace single held-open reader with parallel span reads for content retrieval

cmd/run/contentprovider.go

@@ -1,8 +1,11 @@
 package run
 
 import (
+	"time"
+
 	"github.com/cockroachdb/errors"
 	"github.com/data-preservation-programs/singularity/service/contentprovider"
+	"github.com/data-preservation-programs/singularity/store"
 	"github.com/urfave/cli/v2"
 )
 
@@ -36,6 +39,37 @@ var ContentProviderCmd = &cli.Command{
 			Usage:    "Enable trustless IPFS gateway on /ipfs/",
 			Value:    true,
 		},
+		&cli.IntFlag{
+			Category: "HTTP IPFS Gateway",
+			Name:     "ipfs-span-blocks",
+			Usage:    "Blocks (~1MiB each) fetched per backend read",
+			Value:    8,
+		},
+		&cli.IntFlag{
+			Category: "HTTP IPFS Gateway",
+			Name:     "ipfs-prefetch-spans",
+			Usage:    "Spans prefetched ahead on sequential access",
+			Value:    2,
+		},
+		&cli.IntFlag{
+			Category: "HTTP IPFS Gateway",
+			Name:     "ipfs-max-backend-reads",
+			Usage:    "Max concurrent source storage reads across all requests",
+			Value:    64,
+		},
+		&cli.IntFlag{
+			Category:    "HTTP IPFS Gateway",
+			Name:        "ipfs-cache-blocks",
+			Usage:       "Block cache capacity in blocks (~1MiB each); 0 sizes it to max-backend-reads * span-blocks * (1 + prefetch-spans)",
+			Value:       0,
+			DefaultText: "derived",
+		},
+		&cli.DurationFlag{
+			Category: "HTTP IPFS Gateway",
+			Name:     "ipfs-read-timeout",
+			Usage:    "Max duration of a single backend span read once it holds a connection slot",
+			Value:    2 * time.Minute,
+		},
 	},
 	Action: func(c *cli.Context) error {
 		db, closer, err := openAndMigrate(c)
@@ -49,7 +83,14 @@ var ContentProviderCmd = &cli.Command{
 				EnablePiece:         c.Bool("enable-http-piece"),
 				EnablePieceMetadata: c.Bool("enable-http-piece-metadata"),
 				EnableIPFS:          c.Bool("enable-http-ipfs"),
-				Bind:                c.String("http-bind"),
+				IPFSSpan: store.SpanConfig{
+					SpanBlocks:      c.Int("ipfs-span-blocks"),
+					PrefetchSpans:   c.Int("ipfs-prefetch-spans"),
+					MaxBackendReads: c.Int("ipfs-max-backend-reads"),
+					CacheBlocks:     c.Int("ipfs-cache-blocks"),
+					ReadTimeout:     c.Duration("ipfs-read-timeout"),
+				},
+				Bind: c.String("http-bind"),
 			},
 		}
 

go.mod

@@ -405,7 +405,7 @@ require (
 	golang.org/x/mod v0.32.0 // indirect
 	golang.org/x/net v0.51.0 // indirect
 	golang.org/x/oauth2 v0.33.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sync v0.19.0
 	golang.org/x/sys v0.41.0 // indirect
 	golang.org/x/telemetry v0.0.0-20260109210033-bd525da824e2 // indirect
 	golang.org/x/term v0.40.0 // indirect

model/migrate.go

@@ -88,6 +88,14 @@ func AutoMigrate(db *gorm.DB) error {
 		return errors.Wrap(err, "failed to migrate FK constraints")
 	}
 
+	// Drop the single-column file_id index on car_blocks -- superseded by the
+	// idx_car_blocks_file_span (file_id, file_offset) prefix. Runs after
+	// AutoMigrate so the composite exists before the old one goes away.
+	// GORM never drops indexes on its own, so existing installs need this.
+	if err := db.Exec(`DROP INDEX IF EXISTS idx_car_blocks_file_id`).Error; err != nil {
+		return errors.Wrap(err, "failed to drop redundant car_blocks file_id index")
+	}
+
 	logger.Debug("Creating instance id")
 	err = db.Clauses(clause.OnConflict{
 		DoNothing: true,

model/preparation.go

@@ -296,15 +296,15 @@ type CarBlock struct {
 	CarBlockLength int32      `cbor:"3,keyasint,omitempty" json:"carBlockLength"`                                                       // Length of the block in the Car, including varint, CID and raw block
 	Varint         []byte     `cbor:"4,keyasint,omitempty" json:"varint"`                                                               // Varint is the varint that represents the length of the block and the CID.
 	RawBlock       []byte     `cbor:"5,keyasint,omitempty" json:"rawBlock"`                                                             // Raw block
-	FileOffset     int64      `cbor:"6,keyasint,omitempty" json:"fileOffset"`                                                           // Offset of the block in the File
+	FileOffset     int64      `cbor:"6,keyasint,omitempty" gorm:"index:idx_car_blocks_file_span,priority:2" json:"fileOffset"`          // Offset of the block in the File
 
 	// Internal Caching
 	blockLength int32 // Block length in bytes
 
 	// Associations - SET NULL for fast prep deletion, async cleanup
 	CarID  *CarID  `cbor:"-"                    gorm:"index"                                          json:"carId"`
 	Car    *Car    `cbor:"-"                    gorm:"foreignKey:CarID;constraint:OnDelete:SET NULL"  json:"car,omitempty"  swaggerignore:"true"`
-	FileID *FileID `cbor:"7,keyasint,omitempty" gorm:"index"                                         json:"fileId"`
+	FileID *FileID `cbor:"7,keyasint,omitempty" gorm:"index:idx_car_blocks_file_span,priority:1"            json:"fileId"`
 	File   *File   `cbor:"-"                    gorm:"foreignKey:FileID;constraint:OnDelete:SET NULL" json:"file,omitempty" swaggerignore:"true"`
 }
 

service/contentprovider/contentprovider.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/data-preservation-programs/singularity/service"
+	"github.com/data-preservation-programs/singularity/store"
 	logging "github.com/ipfs/go-log/v2"
 	"gorm.io/gorm"
 )
@@ -22,6 +23,7 @@ type HTTPConfig struct {
 	EnablePiece         bool
 	EnablePieceMetadata bool
 	EnableIPFS          bool
+	IPFSSpan            store.SpanConfig
 	Bind                string
 }
 
@@ -35,6 +37,7 @@ func NewService(db *gorm.DB, config Config) (*Service, error) {
 			enablePiece:         config.HTTP.EnablePiece,
 			enablePieceMetadata: config.HTTP.EnablePieceMetadata,
 			enableIPFS:          config.HTTP.EnableIPFS,
+			ipfsSpanConfig:      config.HTTP.IPFSSpan,
 		})
 	}
 

service/contentprovider/http.go

@@ -31,6 +31,7 @@ type HTTPServer struct {
 	enablePiece         bool
 	enablePieceMetadata bool
 	enableIPFS          bool
+	ipfsSpanConfig      store.SpanConfig
 }
 
 func (*HTTPServer) Name() string {
@@ -107,7 +108,7 @@ func (s *HTTPServer) Start(ctx context.Context, exitErr chan<- error) error {
 	}
 	var bs *store.StorageBlockStore
 	if s.enableIPFS {
-		bs = &store.StorageBlockStore{DBNoContext: s.dbNoContext}
+		bs = store.NewStorageBlockStore(s.dbNoContext, s.ipfsSpanConfig)
 		wrapped := &errorMappingBlockStore{inner: bs}
 		exch := offline.Exchange(wrapped)
 		bsvc := blockservice.New(wrapped, exch)

service/contentprovider/http_test.go

@@ -301,7 +301,7 @@ func TestHTTPServerHandler(t *testing.T) {
 }
 
 func makeGatewayHandler(db *gorm.DB) http.Handler {
-	bs := &store.StorageBlockStore{DBNoContext: db}
+	bs := store.NewStorageBlockStore(db, store.SpanConfig{})
 	wrapped := &errorMappingBlockStore{inner: bs}
 	exch := offline.Exchange(wrapped)
 	bsvc := blockservice.New(wrapped, exch)