package main import ( "context" "log" "sync" "github.com/fatih/color" "github.com/ipfs/boxo/blockservice" exchange "github.com/ipfs/boxo/exchange" offline "github.com/ipfs/boxo/exchange/offline" "github.com/ipfs/boxo/ipld/merkledag" blocks "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" ds "github.com/ipfs/go-datastore" dssync "github.com/ipfs/go-datastore/sync" cbor "github.com/ipfs/go-ipld-cbor" format "github.com/ipfs/go-ipld-format" "golang.org/x/xerrors" "github.com/filecoin-project/lotus/api/v0api" "github.com/filecoin-project/lotus/blockstore" "github.com/filecoin-project/lotus/chain/actors/adt" ) // Stores is a collection of the different stores and services that are needed // to deal with the data layer of Filecoin, conveniently interlinked with one // another. type Stores struct { CBORStore cbor.IpldStore ADTStore adt.Store Datastore ds.Batching Blockstore blockstore.Blockstore BlockService blockservice.BlockService Exchange exchange.Interface DAGService format.DAGService } // NewProxyingStores is a set of Stores backed by a proxying Blockstore that // proxies Get requests for unknown CIDs to a Filecoin node, via the // ChainReadObj RPC. func NewProxyingStores(ctx context.Context, api v0api.FullNode) *Stores { ds := dssync.MutexWrap(ds.NewMapDatastore()) bs := &proxyingBlockstore{ ctx: ctx, api: api, Blockstore: blockstore.FromDatastore(ds), } return NewStores(ctx, ds, bs) } // NewStores creates a non-proxying set of Stores. func NewStores(ctx context.Context, ds ds.Batching, bs blockstore.Blockstore) *Stores { var ( cborstore = cbor.NewCborStore(bs) offl = offline.Exchange(bs) blkserv = blockservice.New(bs, offl) dserv = merkledag.NewDAGService(blkserv) ) return &Stores{ CBORStore: cborstore, ADTStore: adt.WrapStore(ctx, cborstore), Datastore: ds, Blockstore: bs, Exchange: offl, BlockService: blkserv, DAGService: dserv, } } // TracingBlockstore is a Blockstore trait that records CIDs that were accessed // through Get. type TracingBlockstore interface { // StartTracing starts tracing CIDs accessed through the this Blockstore. StartTracing() // FinishTracing finishes tracing accessed CIDs, and returns a map of the // CIDs that were traced. FinishTracing() map[cid.Cid]struct{} } // proxyingBlockstore is a Blockstore wrapper that fetches unknown CIDs from // a Filecoin node via JSON-RPC. type proxyingBlockstore struct { ctx context.Context api v0api.FullNode lk sync.Mutex tracing bool traced map[cid.Cid]struct{} blockstore.Blockstore } var _ TracingBlockstore = (*proxyingBlockstore)(nil) func (pb *proxyingBlockstore) StartTracing() { pb.lk.Lock() pb.tracing = true pb.traced = map[cid.Cid]struct{}{} pb.lk.Unlock() } func (pb *proxyingBlockstore) FinishTracing() map[cid.Cid]struct{} { pb.lk.Lock() ret := pb.traced pb.tracing = false pb.traced = map[cid.Cid]struct{}{} pb.lk.Unlock() return ret } func (pb *proxyingBlockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error) { pb.lk.Lock() if pb.tracing { pb.traced[cid] = struct{}{} } pb.lk.Unlock() if block, err := pb.Blockstore.Get(ctx, cid); err == nil { return block, err } log.Println(color.CyanString("fetching cid via rpc: %v", cid)) item, err := pb.api.ChainReadObj(pb.ctx, cid) if err != nil { return nil, err } block, err := blocks.NewBlockWithCid(item, cid) if err != nil { return nil, err } err = pb.Blockstore.Put(ctx, block) if err != nil { return nil, err } return block, nil } func (pb *proxyingBlockstore) Put(ctx context.Context, block blocks.Block) error { pb.lk.Lock() if pb.tracing { pb.traced[block.Cid()] = struct{}{} } pb.lk.Unlock() return pb.Blockstore.Put(ctx, block) } func (pb *proxyingBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error { pb.lk.Lock() if pb.tracing { for _, b := range blocks { pb.traced[b.Cid()] = struct{}{} } } pb.lk.Unlock() return pb.Blockstore.PutMany(ctx, blocks) } func (pb *proxyingBlockstore) View(ctx context.Context, c cid.Cid, callback func([]byte) error) error { blk, err := pb.Get(ctx, c) if err != nil { return xerrors.Errorf("failed to Get cid %s: %w", c, err) } return callback(blk.RawData()) }