diff --git a/cmd/run/contentprovider.go b/cmd/run/contentprovider.go index 01dcf1e3..d97ba1a4 100644 --- a/cmd/run/contentprovider.go +++ b/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"), }, } diff --git a/go.mod b/go.mod index 6f586f3e..bd4887e3 100644 --- a/go.mod +++ b/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 diff --git a/model/migrate.go b/model/migrate.go index 5b415cb9..78f75d18 100644 --- a/model/migrate.go +++ b/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, diff --git a/model/preparation.go b/model/preparation.go index fd19ab1a..f35d12b5 100644 --- a/model/preparation.go +++ b/model/preparation.go @@ -296,7 +296,7 @@ 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 @@ -304,7 +304,7 @@ type CarBlock struct { // 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"` } diff --git a/service/contentprovider/contentprovider.go b/service/contentprovider/contentprovider.go index 199397e2..af73dd5d 100644 --- a/service/contentprovider/contentprovider.go +++ b/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, }) } diff --git a/service/contentprovider/http.go b/service/contentprovider/http.go index 8f742722..a73f7152 100644 --- a/service/contentprovider/http.go +++ b/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) diff --git a/service/contentprovider/http_test.go b/service/contentprovider/http_test.go index d3110d5f..239b3650 100644 --- a/service/contentprovider/http_test.go +++ b/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) diff --git a/store/storage_blockstore.go b/store/storage_blockstore.go index d7c4f4b3..5c5c4319 100644 --- a/store/storage_blockstore.go +++ b/store/storage_blockstore.go @@ -2,53 +2,137 @@ package store import ( "context" + "fmt" "io" "sync" + "time" "github.com/cockroachdb/errors" "github.com/data-preservation-programs/singularity/model" "github.com/data-preservation-programs/singularity/storagesystem" "github.com/data-preservation-programs/singularity/util" + lru "github.com/hashicorp/golang-lru/v2" blocks "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" format "github.com/ipfs/go-ipld-format" + "golang.org/x/sync/singleflight" "gorm.io/gorm" ) +// SpanConfig tunes the span-read behavior of StorageBlockStore. +// Blocks are ~1MiB (pack chunk size), so block counts are roughly MiB. +type SpanConfig struct { + // SpanBlocks is the number of consecutive blocks fetched per backend read. + SpanBlocks int + // PrefetchSpans is the number of spans prefetched ahead once access to a + // file looks sequential. + PrefetchSpans int + // MaxBackendReads caps concurrent backend reads across all requests. + MaxBackendReads int + // CacheBlocks is the block cache capacity in blocks. When <= 0 it is + // derived from the other settings. + CacheBlocks int + // ReadTimeout bounds a single backend span read once it holds a + // connection slot, so a hung backend cannot pin slots indefinitely. + ReadTimeout time.Duration +} + +func (c SpanConfig) withDefaults() SpanConfig { + if c.SpanBlocks <= 0 { + c.SpanBlocks = 8 + } + if c.PrefetchSpans <= 0 { + c.PrefetchSpans = 2 + } + if c.MaxBackendReads <= 0 { + c.MaxBackendReads = 64 + } + // an active sequential client keeps 1+PrefetchSpans spans warm; size the + // cache so MaxBackendReads concurrent streams don't evict each other's + // unconsumed spans and re-fetch them + if c.CacheBlocks <= 0 { + c.CacheBlocks = c.MaxBackendReads * c.SpanBlocks * (1 + c.PrefetchSpans) + } + if c.ReadTimeout <= 0 { + c.ReadTimeout = 2 * time.Minute + } + return c +} + // StorageBlockStore is a blockstore backed by the singularity database and -// rclone storage backends. It pools rclone handlers per storage and holds -// open a streaming file reader to serve sequential block reads efficiently. +// rclone storage backends. // // DAG nodes (directory structure, file roots) are stored inline in the DB // and returned without any storage I/O. File-backed leaf blocks are read -// from source files via rclone, with a single-entry reader cache that -// exploits the depth-first traversal pattern: consecutive Get() calls for -// blocks from the same file read from the same held-open stream. +// from source files via ranged reads covering a span of consecutive blocks; +// spans land in a bounded LRU cache so the sequential Gets that follow are +// served from memory. Sequential access additionally prefetches the next +// spans in parallel, which is what provides per-client throughput beyond +// the backend's per-connection rate. There are no held-open streams and no +// global lock: concurrency is bounded only by the backend-read semaphore. type StorageBlockStore struct { - DBNoContext *gorm.DB + dbNoContext *gorm.DB + cfg SpanConfig - mu sync.Mutex - handlers map[model.StorageID]*storagesystem.RCloneHandler - active *fileReader -} + handlersMu sync.RWMutex + handlers map[model.StorageID]*storagesystem.RCloneHandler -type fileReader struct { - fileID model.FileID - reader io.ReadCloser - offset int64 + cache *lru.Cache[string, []byte] + inflight singleflight.Group + sem chan struct{} + + // seq tracks, per file, the offset window in which the next cache miss + // counts as a sequential continuation. Entries are disposable hints. + seqMu sync.Mutex + seq map[model.FileID]seqWindow + + // bg outlives individual requests: span results are shared via + // singleflight and cached, so a fetch must not die with the request + // that happened to trigger it. Close cancels it. + bg context.Context cancel context.CancelFunc } +type seqWindow struct { + lo, hi int64 +} + +func NewStorageBlockStore(db *gorm.DB, cfg SpanConfig) *StorageBlockStore { + cfg = cfg.withDefaults() + cache, err := lru.New[string, []byte](cfg.CacheBlocks) + if err != nil { + panic(err) + } + bg, cancel := context.WithCancel(context.Background()) + return &StorageBlockStore{ + dbNoContext: db, + cfg: cfg, + handlers: make(map[model.StorageID]*storagesystem.RCloneHandler), + cache: cache, + sem: make(chan struct{}, cfg.MaxBackendReads), + seq: make(map[model.FileID]seqWindow), + bg: bg, + cancel: cancel, + } +} + func (s *StorageBlockStore) Has(ctx context.Context, c cid.Cid) (bool, error) { + if s.cache.Contains(c.KeyString()) { + return true, nil + } var count int64 - err := s.DBNoContext.WithContext(ctx).Model(&model.CarBlock{}). + err := s.dbNoContext.WithContext(ctx).Model(&model.CarBlock{}). Select("cid").Where("cid = ?", model.CID(c)).Count(&count).Error return count > 0, errors.WithStack(err) } func (s *StorageBlockStore) Get(ctx context.Context, c cid.Cid) (blocks.Block, error) { + if data, ok := s.cache.Get(c.KeyString()); ok { + return blocks.NewBlockWithCid(data, c) + } + var carBlock model.CarBlock - err := s.DBNoContext.WithContext(ctx). + err := s.dbNoContext.WithContext(ctx). Joins("File.Attachment.Storage"). Where("car_blocks.cid = ?", model.CID(c)). First(&carBlock).Error @@ -68,91 +152,222 @@ func (s *StorageBlockStore) Get(ctx context.Context, c cid.Cid) (blocks.Block, e } func (s *StorageBlockStore) readFileBlock(ctx context.Context, carBlock model.CarBlock, c cid.Cid) (blocks.Block, error) { - s.mu.Lock() - defer s.mu.Unlock() - if carBlock.File == nil || carBlock.File.Attachment == nil || carBlock.File.Attachment.Storage == nil { return nil, errors.Errorf("block %s has no associated storage (orphaned or deleted source file)", c) } - storage := *carBlock.File.Attachment.Storage file := *carBlock.File - blockLen := int64(carBlock.BlockLength()) + storage := *carBlock.File.Attachment.Storage - handler, err := s.getHandler(ctx, storage) + blks, end, err := s.fetchSpan(ctx, file, storage, carBlock.FileOffset) if err != nil { - return nil, errors.WithStack(err) + return nil, err + } + + s.maybePrefetch(file, storage, carBlock.FileOffset, end) + + for _, blk := range blks { + if blk.Cid().Equals(c) { + return blk, nil + } } + return nil, errors.Errorf("block %s missing from fetched span (car_blocks changed underneath)", c) +} + +type spanResult struct { + blks []blocks.Block + end int64 +} - // try to read from held-open stream - if s.active != nil && s.active.fileID == file.ID && s.active.offset == carBlock.FileOffset { - data, err := s.readFromActive(blockLen) +// fetchSpan reads one span of consecutive blocks starting at offset and +// populates the cache. Concurrent fetches of the same span are collapsed. +// The fetch itself runs on the store's background context: the result is +// shared, so it must not be poisoned by one caller's cancellation. Only the +// wait is bound to ctx -- a caller whose client disconnected detaches while +// the fetch continues and still lands in the cache. +func (s *StorageBlockStore) fetchSpan(ctx context.Context, file model.File, storage model.Storage, offset int64) ([]blocks.Block, int64, error) { + key := fmt.Sprintf("%d:%d", file.ID, offset) + ch := s.inflight.DoChan(key, func() (any, error) { + rows, err := s.spanRows(file.ID, offset, s.cfg.SpanBlocks) if err != nil { - s.closeActive() return nil, err } - return blocks.NewBlockWithCid(data, c) + if len(rows) == 0 { + return &spanResult{}, nil + } + blks, err := s.readRows(file, storage, rows) + if err != nil { + return nil, err + } + last := rows[len(rows)-1] + return &spanResult{blks: blks, end: last.FileOffset + int64(last.BlockLength())}, nil + }) + select { + case res := <-ch: + if res.Err != nil { + return nil, 0, res.Err + } + r, _ := res.Val.(*spanResult) + return r.blks, r.end, nil + case <-ctx.Done(): + return nil, 0, errors.WithStack(ctx.Err()) } +} - // different file or non-sequential offset -- close old, open new - s.closeActive() - - // use a detached context for the cached reader so it outlives the - // request that created it. closeActive() cancels it on cleanup. - readerCtx, readerCancel := context.WithCancel(context.Background()) - reader, obj, err := handler.Read(readerCtx, file.Path, carBlock.FileOffset, file.Size-carBlock.FileOffset) +// spanRows returns up to n car_blocks rows for a file starting at offset, +// trimmed to the contiguous run so a single ranged read covers them exactly. +func (s *StorageBlockStore) spanRows(fileID model.FileID, offset int64, n int) ([]model.CarBlock, error) { + var rows []model.CarBlock + err := s.dbNoContext.WithContext(s.bg). + Where("file_id = ? AND file_offset >= ?", fileID, offset). + Order("file_offset").Limit(n).Find(&rows).Error if err != nil { - readerCancel() return nil, errors.WithStack(err) } - - same, explanation := storagesystem.IsSameEntry(ctx, file, obj) - if !same { - reader.Close() - readerCancel() - return nil, errors.Wrap(ErrFileHasChanged, explanation) + end := int64(-1) + for i := range rows { + if i > 0 && rows[i].FileOffset != end { + rows = rows[:i] + break + } + end = rows[i].FileOffset + int64(rows[i].BlockLength()) } + return rows, nil +} - s.active = &fileReader{ - fileID: file.ID, - reader: reader, - offset: carBlock.FileOffset, - cancel: readerCancel, +// readRows performs the ranged backend read covering rows and caches the +// resulting blocks. Each block owns its own allocation so a retained block +// never pins the rest of the span. +func (s *StorageBlockStore) readRows(file model.File, storage model.Storage, rows []model.CarBlock) ([]blocks.Block, error) { + blks := make([]blocks.Block, len(rows)) + + // a prefetch may race a sync read of an overlapping span after a window + // reset -- if everything is already cached, skip the backend read + allCached := true + for i := range rows { + data, ok := s.cache.Get(cid.Cid(rows[i].CID).KeyString()) + if !ok { + allCached = false + break + } + blk, err := blocks.NewBlockWithCid(data, cid.Cid(rows[i].CID)) + if err != nil { + return nil, errors.WithStack(err) + } + blks[i] = blk + } + if allCached { + return blks, nil } - data, err := s.readFromActive(blockLen) + handler, err := s.getHandler(storage) if err != nil { - s.closeActive() return nil, err } - return blocks.NewBlockWithCid(data, c) -} -func (s *StorageBlockStore) readFromActive(length int64) ([]byte, error) { - buf := make([]byte, length) - _, err := io.ReadFull(s.active.reader, buf) + var total int64 + for i := range rows { + total += int64(rows[i].BlockLength()) + } + + select { + case s.sem <- struct{}{}: + case <-s.bg.Done(): + return nil, errors.WithStack(s.bg.Err()) + } + defer func() { <-s.sem }() + + // bound slot occupancy -- a hung backend read must not pin a + // connection slot until shutdown + rctx, rcancel := context.WithTimeout(s.bg, s.cfg.ReadTimeout) + defer rcancel() + + reader, obj, err := handler.Read(rctx, file.Path, rows[0].FileOffset, total) if err != nil { return nil, errors.WithStack(err) } - s.active.offset += length - return buf, nil + defer reader.Close() + + same, explanation := storagesystem.IsSameEntry(rctx, file, obj) + if !same { + return nil, errors.Wrap(ErrFileHasChanged, explanation) + } + + for i := range rows { + data := make([]byte, rows[i].BlockLength()) + if _, err := io.ReadFull(reader, data); err != nil { + return nil, errors.WithStack(err) + } + blk, err := blocks.NewBlockWithCid(data, cid.Cid(rows[i].CID)) + if err != nil { + return nil, errors.WithStack(err) + } + blks[i] = blk + s.cache.Add(cid.Cid(rows[i].CID).KeyString(), data) + } + return blks, nil } -func (s *StorageBlockStore) closeActive() { - if s.active != nil { - s.active.reader.Close() - s.active.cancel() - s.active = nil +// maybePrefetch schedules background reads of the next spans when access +// looks sequential. The first span of a file only opens the window; the +// second consecutive one starts prefetching, so random seeks and small +// files never trigger it. +func (s *StorageBlockStore) maybePrefetch(file model.File, storage model.Storage, offset, end int64) { + if end <= 0 { + return + } + + s.seqMu.Lock() + w, ok := s.seq[file.ID] + sequential := ok && offset > w.lo && offset <= w.hi + // hints are disposable -- reset rather than track LRU + if len(s.seq) > 4096 { + s.seq = make(map[model.FileID]seqWindow) } + s.seq[file.ID] = seqWindow{lo: offset, hi: end} + s.seqMu.Unlock() + + if !sequential { + return + } + + go func() { + rows, err := s.spanRows(file.ID, end, s.cfg.SpanBlocks*s.cfg.PrefetchSpans) + if err != nil || len(rows) == 0 { + return + } + // spans fetch in parallel -- this is the multipart read + for i := 0; i < len(rows); i += s.cfg.SpanBlocks { + go func(off int64) { + _, _, _ = s.fetchSpan(s.bg, file, storage, off) + }(rows[i].FileOffset) + } + // extend the window past the prefetched region so the miss that + // follows its consumption still counts as sequential + last := rows[len(rows)-1] + hi := last.FileOffset + int64(last.BlockLength()) + s.seqMu.Lock() + if w, ok := s.seq[file.ID]; ok && hi > w.hi { + w.hi = hi + s.seq[file.ID] = w + } + s.seqMu.Unlock() + }() } -func (s *StorageBlockStore) getHandler(ctx context.Context, storage model.Storage) (*storagesystem.RCloneHandler, error) { - if s.handlers == nil { - s.handlers = make(map[model.StorageID]*storagesystem.RCloneHandler) +func (s *StorageBlockStore) getHandler(storage model.Storage) (*storagesystem.RCloneHandler, error) { + s.handlersMu.RLock() + h, ok := s.handlers[storage.ID] + s.handlersMu.RUnlock() + if ok { + return h, nil } + + s.handlersMu.Lock() + defer s.handlersMu.Unlock() if h, ok := s.handlers[storage.ID]; ok { return h, nil } - h, err := storagesystem.NewRCloneHandler(ctx, storage) + h, err := storagesystem.NewRCloneHandler(s.bg, storage) if err != nil { return nil, err } @@ -161,8 +376,11 @@ func (s *StorageBlockStore) getHandler(ctx context.Context, storage model.Storag } func (s *StorageBlockStore) GetSize(ctx context.Context, c cid.Cid) (int, error) { + if data, ok := s.cache.Get(c.KeyString()); ok { + return len(data), nil + } var carBlock model.CarBlock - err := s.DBNoContext.WithContext(ctx).Where("cid = ?", model.CID(c)).First(&carBlock).Error + err := s.dbNoContext.WithContext(ctx).Where("cid = ?", model.CID(c)).First(&carBlock).Error if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return 0, format.ErrNotFound{Cid: c} @@ -190,11 +408,10 @@ func (s *StorageBlockStore) DeleteBlock(ctx context.Context, c cid.Cid) error { return util.ErrNotImplemented } -// Close releases all held resources. -// rclone handler cleanup is not implemented -- backends may hold -// connections (e.g. SFTP) but RCloneHandler doesn't expose Shutdown. +// Close cancels in-flight background reads; they exit on context +// cancellation. rclone handler cleanup is not implemented -- backends may +// hold connections (e.g. SFTP) but RCloneHandler doesn't expose Shutdown. func (s *StorageBlockStore) Close() { - s.mu.Lock() - defer s.mu.Unlock() - s.closeActive() + s.cancel() + s.cache.Purge() } diff --git a/store/storage_blockstore_test.go b/store/storage_blockstore_test.go new file mode 100644 index 00000000..a3cf458c --- /dev/null +++ b/store/storage_blockstore_test.go @@ -0,0 +1,234 @@ +package store + +import ( + "context" + "math/rand" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/data-preservation-programs/singularity/model" + "github.com/data-preservation-programs/singularity/util/testutil" + "github.com/ipfs/boxo/util" + "github.com/ipfs/go-cid" + format "github.com/ipfs/go-ipld-format" + "github.com/multiformats/go-varint" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +// makeSpanFixture writes a file to a local storage and creates car_blocks +// leaf rows slicing it into blockSize chunks, mirroring what pack produces. +func makeSpanFixture(t *testing.T, db *gorm.DB, content []byte, blockSize int) []cid.Cid { + t.Helper() + tmp := t.TempDir() + name := "data.bin" + require.NoError(t, os.WriteFile(filepath.Join(tmp, name), content, 0644)) + + prep := &model.Preparation{Name: t.Name() + filepath.Base(tmp)} + require.NoError(t, db.Create(prep).Error) + storage := &model.Storage{Name: prep.Name, Type: "local", Path: tmp} + require.NoError(t, db.Create(storage).Error) + attachment := &model.SourceAttachment{PreparationID: prep.ID, StorageID: storage.ID} + require.NoError(t, db.Create(attachment).Error) + dir := &model.Directory{AttachmentID: &attachment.ID} + require.NoError(t, db.Create(dir).Error) + file := &model.File{ + Path: name, + Size: int64(len(content)), + LastModifiedNano: testutil.GetFileTimestamp(t, filepath.Join(tmp, name)), + AttachmentID: &attachment.ID, + DirectoryID: &dir.ID, + } + require.NoError(t, db.Create(file).Error) + car := &model.Car{ + PieceCID: model.CID(cid.NewCidV1(cid.FilCommitmentUnsealed, util.Hash([]byte(t.Name())))), + PieceSize: 1 << 20, + RootCID: model.CID(testutil.TestCid), + PreparationID: &prep.ID, + AttachmentID: &attachment.ID, + PieceType: model.DataPiece, + } + require.NoError(t, db.Create(car).Error) + + var cids []cid.Cid + carOffset := int64(59) + for off := 0; off < len(content); off += blockSize { + chunk := content[off:min(off+blockSize, len(content))] + c := cid.NewCidV1(cid.Raw, util.Hash(chunk)) + v := varint.ToUvarint(uint64(c.ByteLen() + len(chunk))) + require.NoError(t, db.Create(&model.CarBlock{ + CarID: &car.ID, + CID: model.CID(c), + CarOffset: carOffset, + CarBlockLength: int32(len(v) + c.ByteLen() + len(chunk)), + Varint: v, + FileID: &file.ID, + FileOffset: int64(off), + }).Error) + carOffset += int64(len(v) + c.ByteLen() + len(chunk)) + cids = append(cids, c) + } + return cids +} + +func TestStorageBlockStore_SequentialRead(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + content := testutil.GenerateRandomBytes(100) + blockSize := 8 + cids := makeSpanFixture(t, db, content, blockSize) + + s := NewStorageBlockStore(db, SpanConfig{SpanBlocks: 3, PrefetchSpans: 2, CacheBlocks: 64}) + defer s.Close() + + for i, c := range cids { + blk, err := s.Get(ctx, c) + require.NoError(t, err) + expected := content[i*blockSize : min((i+1)*blockSize, len(content))] + require.Equal(t, expected, blk.RawData()) + + size, err := s.GetSize(ctx, c) + require.NoError(t, err) + require.Equal(t, len(expected), size) + } + }) +} + +func TestStorageBlockStore_RandomAccess(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + content := testutil.GenerateRandomBytes(96) + blockSize := 8 + cids := makeSpanFixture(t, db, content, blockSize) + + s := NewStorageBlockStore(db, SpanConfig{SpanBlocks: 2, PrefetchSpans: 1, CacheBlocks: 64}) + defer s.Close() + + for i := len(cids) - 1; i >= 0; i-- { + blk, err := s.Get(ctx, cids[i]) + require.NoError(t, err) + require.Equal(t, content[i*blockSize:(i+1)*blockSize], blk.RawData()) + } + }) +} + +func TestStorageBlockStore_CacheServesWithoutDB(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + content := testutil.GenerateRandomBytes(32) + cids := makeSpanFixture(t, db, content, 8) + + s := NewStorageBlockStore(db, SpanConfig{SpanBlocks: 4, CacheBlocks: 64}) + defer s.Close() + + // first Get fetches the whole span into cache + _, err := s.Get(ctx, cids[0]) + require.NoError(t, err) + + require.NoError(t, db.Where("1 = 1").Delete(&model.CarBlock{}).Error) + + // all blocks of the span are still served from cache + for i, c := range cids { + blk, err := s.Get(ctx, c) + require.NoError(t, err) + require.Equal(t, content[i*8:(i+1)*8], blk.RawData()) + + has, err := s.Has(ctx, c) + require.NoError(t, err) + require.True(t, has) + } + }) +} + +func TestStorageBlockStore_ConcurrentReads(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + contentA := testutil.GenerateRandomBytes(128) + contentB := testutil.GenerateRandomBytes(128) + cidsA := makeSpanFixture(t, db, contentA, 8) + cidsB := makeSpanFixture(t, db, contentB, 8) + + s := NewStorageBlockStore(db, SpanConfig{SpanBlocks: 2, PrefetchSpans: 2, MaxBackendReads: 4, CacheBlocks: 8}) + defer s.Close() + + var wg sync.WaitGroup + for g := 0; g < 8; g++ { + wg.Add(1) + go func(seed int64) { + defer wg.Done() + r := rand.New(rand.NewSource(seed)) + for range 50 { + content, cids := contentA, cidsA + if r.Intn(2) == 1 { + content, cids = contentB, cidsB + } + i := r.Intn(len(cids)) + blk, err := s.Get(ctx, cids[i]) + require.NoError(t, err) + require.Equal(t, content[i*8:(i+1)*8], blk.RawData()) + } + }(int64(g)) + } + wg.Wait() + }) +} + +func TestStorageBlockStore_FileChanged(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + content := testutil.GenerateRandomBytes(32) + cids := makeSpanFixture(t, db, content, 8) + + var storage model.Storage + require.NoError(t, db.First(&storage).Error) + require.NoError(t, os.WriteFile(filepath.Join(storage.Path, "data.bin"), []byte("changed"), 0644)) + + s := NewStorageBlockStore(db, SpanConfig{}) + defer s.Close() + + _, err := s.Get(ctx, cids[0]) + require.ErrorIs(t, err, ErrFileHasChanged) + }) +} + +func TestStorageBlockStore_NotFound(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + s := NewStorageBlockStore(db, SpanConfig{}) + defer s.Close() + + _, err := s.Get(ctx, testutil.TestCid) + require.ErrorAs(t, err, &format.ErrNotFound{}) + _, err = s.GetSize(ctx, testutil.TestCid) + require.ErrorAs(t, err, &format.ErrNotFound{}) + has, err := s.Has(ctx, testutil.TestCid) + require.NoError(t, err) + require.False(t, has) + }) +} + +func TestStorageBlockStore_CancelledWaiterDetaches(t *testing.T) { + testutil.All(t, func(ctx context.Context, t *testing.T, db *gorm.DB) { + content := testutil.GenerateRandomBytes(32) + cids := makeSpanFixture(t, db, content, 8) + + s := NewStorageBlockStore(db, SpanConfig{SpanBlocks: 4, MaxBackendReads: 1}) + defer s.Close() + + // occupy the only connection slot so the fetch parks on the semaphore + s.sem <- struct{}{} + + waitCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + _, err := s.Get(waitCtx, cids[0]) + require.ErrorIs(t, err, context.DeadlineExceeded) + + // the fetch survived the caller: free the slot and it completes into + // the cache + <-s.sem + require.Eventually(t, func() bool { + return s.cache.Contains(cids[0].KeyString()) + }, 5*time.Second, 10*time.Millisecond) + + blk, err := s.Get(ctx, cids[0]) + require.NoError(t, err) + require.Equal(t, content[:8], blk.RawData()) + }) +}