| 1 | package node |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | |
| 8 | "github.com/ipfs/boxo/blockservice" |
| 9 | blockstore "github.com/ipfs/boxo/blockstore" |
| 10 | exchange "github.com/ipfs/boxo/exchange" |
| 11 | offline "github.com/ipfs/boxo/exchange/offline" |
| 12 | "github.com/ipfs/boxo/fetcher" |
| 13 | bsfetcher "github.com/ipfs/boxo/fetcher/impl/blockservice" |
| 14 | "github.com/ipfs/boxo/filestore" |
| 15 | "github.com/ipfs/boxo/ipld/merkledag" |
| 16 | "github.com/ipfs/boxo/ipld/unixfs" |
| 17 | "github.com/ipfs/boxo/mfs" |
| 18 | pathresolver "github.com/ipfs/boxo/path/resolver" |
| 19 | pin "github.com/ipfs/boxo/pinning/pinner" |
| 20 | "github.com/ipfs/boxo/pinning/pinner/dspinner" |
| 21 | "github.com/ipfs/go-cid" |
| 22 | "github.com/ipfs/go-datastore" |
| 23 | format "github.com/ipfs/go-ipld-format" |
| 24 | "github.com/ipfs/go-unixfsnode" |
| 25 | dagpb "github.com/ipld/go-codec-dagpb" |
| 26 | "go.uber.org/fx" |
| 27 | |
| 28 | "github.com/ipfs/kubo/config" |
| 29 | "github.com/ipfs/kubo/core/node/helpers" |
| 30 | "github.com/ipfs/kubo/core/shutdown" |
| 31 | "github.com/ipfs/kubo/repo" |
| 32 | ) |
| 33 | |
| 34 | // FilesRootDatastoreKey is the datastore key for the MFS files root CID. |
| 35 | var FilesRootDatastoreKey = datastore.NewKey("/local/filesroot") |
| 36 | |
| 37 | // BlockService creates new blockservice which provides an interface to fetch content-addressable blocks |
| 38 | func BlockService(cfg *config.Config) func(lc fx.Lifecycle, bs blockstore.Blockstore, rem exchange.Interface) blockservice.BlockService { |
| 39 | return func(lc fx.Lifecycle, bs blockstore.Blockstore, rem exchange.Interface) blockservice.BlockService { |
| 40 | bsvc := blockservice.New(bs, rem, |
| 41 | blockservice.WriteThrough(cfg.Datastore.WriteThrough.WithDefault(config.DefaultWriteThrough)), |
| 42 | ) |
| 43 | |
| 44 | lc.Append(fx.Hook{ |
| 45 | OnStop: func(ctx context.Context) error { |
| 46 | return shutdown.CloseWithCtx(ctx, "blockservice", bsvc.Close) |
| 47 | }, |
| 48 | }) |
| 49 | |
| 50 | return bsvc |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | // Pinning builds the pinner that GC uses to decide which blocks to keep. |
| 55 | // |
| 56 | // An fx OnStop hook closes the pinner before the repo (and its |
| 57 | // datastore). The order matters: in-flight pinner operations hold a |
| 58 | // reference to the datastore, and some datastores (pebble) panic on |
| 59 | // use after Close. Pinner.Close cancels those operations and waits |
| 60 | // for them to return. See |
| 61 | // [github.com/ipfs/boxo/pinning/pinner.Pinner.Close]. |
| 62 | func Pinning(strategy string) func(lc fx.Lifecycle, bstore blockstore.Blockstore, ds format.DAGService, repo repo.Repo, prov DHTProvider) (pin.Pinner, error) { |
| 63 | strategyFlag := config.MustParseProvideStrategy(strategy) |
| 64 | |
| 65 | return func(lc fx.Lifecycle, |
| 66 | bstore blockstore.Blockstore, |
| 67 | ds format.DAGService, |
| 68 | repo repo.Repo, |
| 69 | prov DHTProvider, |
| 70 | ) (pin.Pinner, error) { |
| 71 | rootDS := repo.Datastore() |
| 72 | |
| 73 | syncFn := func(ctx context.Context) error { |
| 74 | if err := rootDS.Sync(ctx, blockstore.BlockPrefix); err != nil { |
| 75 | return err |
| 76 | } |
| 77 | return rootDS.Sync(ctx, filestore.FilestorePrefix) |
| 78 | } |
| 79 | syncDs := &syncDagService{ds, syncFn} |
| 80 | |
| 81 | ctx := context.TODO() |
| 82 | |
| 83 | var opts []dspinner.Option |
| 84 | roots := (strategyFlag & config.ProvideStrategyRoots) != 0 |
| 85 | pinned := (strategyFlag & config.ProvideStrategyPinned) != 0 |
| 86 | |
| 87 | // Important: Only one of WithPinnedProvider or WithRootsProvider should be active. |
| 88 | // Having both would cause duplicate root advertisements since "pinned" includes all |
| 89 | // pinned content (roots + children), while "roots" is just the root CIDs. |
| 90 | // We prioritize "pinned" if both are somehow set (though this shouldn't happen |
| 91 | // with proper strategy parsing). |
| 92 | if pinned { |
| 93 | opts = append(opts, dspinner.WithPinnedProvider(prov)) |
| 94 | } else if roots { |
| 95 | opts = append(opts, dspinner.WithRootsProvider(prov)) |
| 96 | } |
| 97 | |
| 98 | pinning, err := dspinner.New(ctx, rootDS, syncDs, opts...) |
| 99 | if err != nil { |
| 100 | return nil, err |
| 101 | } |
| 102 | |
| 103 | // fx runs OnStop hooks in reverse registration order. The |
| 104 | // repo provider registers its close hook earlier (in |
| 105 | // builder.go), so this hook runs first and the repo hook |
| 106 | // runs after, without an explicit dependency between them. |
| 107 | // |
| 108 | // Wrapped with CloseWithCtx because the boxo Pinner.Close |
| 109 | // contract notes that an in-flight op which ignores its ctx |
| 110 | // (a downstream bug) can block Close; the host must bound it |
| 111 | // at the call site so the shutdown deadline is honored. |
| 112 | lc.Append(fx.Hook{ |
| 113 | OnStop: func(ctx context.Context) error { |
| 114 | return shutdown.CloseWithCtx(ctx, "pinner", pinning.Close) |
| 115 | }, |
| 116 | }) |
| 117 | |
| 118 | return pinning, nil |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | var ( |
| 123 | _ merkledag.SessionMaker = new(syncDagService) |
| 124 | _ format.DAGService = new(syncDagService) |
| 125 | ) |
| 126 | |
| 127 | // syncDagService is used by the Pinner to ensure data gets persisted to the underlying datastore |
| 128 | type syncDagService struct { |
| 129 | format.DAGService |
| 130 | syncFn func(context.Context) error |
| 131 | } |
| 132 | |
| 133 | func (s *syncDagService) Sync(ctx context.Context) error { |
| 134 | return s.syncFn(ctx) |
| 135 | } |
| 136 | |
| 137 | func (s *syncDagService) Session(ctx context.Context) format.NodeGetter { |
| 138 | return merkledag.NewSession(ctx, s.DAGService) |
| 139 | } |
| 140 | |
| 141 | // FetchersOut allows injection of fetchers. |
| 142 | type FetchersOut struct { |
| 143 | fx.Out |
| 144 | IPLDFetcher fetcher.Factory `name:"ipldFetcher"` |
| 145 | UnixfsFetcher fetcher.Factory `name:"unixfsFetcher"` |
| 146 | OfflineIPLDFetcher fetcher.Factory `name:"offlineIpldFetcher"` |
| 147 | OfflineUnixfsFetcher fetcher.Factory `name:"offlineUnixfsFetcher"` |
| 148 | } |
| 149 | |
| 150 | // FetchersIn allows using fetchers for other dependencies. |
| 151 | type FetchersIn struct { |
| 152 | fx.In |
| 153 | IPLDFetcher fetcher.Factory `name:"ipldFetcher"` |
| 154 | UnixfsFetcher fetcher.Factory `name:"unixfsFetcher"` |
| 155 | OfflineIPLDFetcher fetcher.Factory `name:"offlineIpldFetcher"` |
| 156 | OfflineUnixfsFetcher fetcher.Factory `name:"offlineUnixfsFetcher"` |
| 157 | } |
| 158 | |
| 159 | // FetcherConfig returns a fetcher config that can build new fetcher instances |
| 160 | func FetcherConfig(bs blockservice.BlockService) FetchersOut { |
| 161 | ipldFetcher := bsfetcher.NewFetcherConfig(bs) |
| 162 | ipldFetcher.PrototypeChooser = dagpb.AddSupportToChooser(bsfetcher.DefaultPrototypeChooser) |
| 163 | unixFSFetcher := ipldFetcher.WithReifier(unixfsnode.Reify) |
| 164 | |
| 165 | // Construct offline versions which we can safely use in contexts where |
| 166 | // path resolution should not fetch new blocks via exchange. |
| 167 | offlineBs := blockservice.New(bs.Blockstore(), offline.Exchange(bs.Blockstore())) |
| 168 | offlineIpldFetcher := bsfetcher.NewFetcherConfig(offlineBs) |
| 169 | offlineIpldFetcher.SkipNotFound = true // carries onto the UnixFSFetcher below |
| 170 | offlineIpldFetcher.PrototypeChooser = dagpb.AddSupportToChooser(bsfetcher.DefaultPrototypeChooser) |
| 171 | offlineUnixFSFetcher := offlineIpldFetcher.WithReifier(unixfsnode.Reify) |
| 172 | |
| 173 | return FetchersOut{ |
| 174 | IPLDFetcher: ipldFetcher, |
| 175 | UnixfsFetcher: unixFSFetcher, |
| 176 | OfflineIPLDFetcher: offlineIpldFetcher, |
| 177 | OfflineUnixfsFetcher: offlineUnixFSFetcher, |
| 178 | } |
| 179 | } |
| 180 | |
| 181 | // PathResolversOut allows injection of path resolvers |
| 182 | type PathResolversOut struct { |
| 183 | fx.Out |
| 184 | IPLDPathResolver pathresolver.Resolver `name:"ipldPathResolver"` |
| 185 | UnixFSPathResolver pathresolver.Resolver `name:"unixFSPathResolver"` |
| 186 | OfflineIPLDPathResolver pathresolver.Resolver `name:"offlineIpldPathResolver"` |
| 187 | OfflineUnixFSPathResolver pathresolver.Resolver `name:"offlineUnixFSPathResolver"` |
| 188 | } |
| 189 | |
| 190 | // PathResolverConfig creates path resolvers with the given fetchers. |
| 191 | func PathResolverConfig(fetchers FetchersIn) PathResolversOut { |
| 192 | return PathResolversOut{ |
| 193 | IPLDPathResolver: pathresolver.NewBasicResolver(fetchers.IPLDFetcher), |
| 194 | UnixFSPathResolver: pathresolver.NewBasicResolver(fetchers.UnixfsFetcher), |
| 195 | OfflineIPLDPathResolver: pathresolver.NewBasicResolver(fetchers.OfflineIPLDFetcher), |
| 196 | OfflineUnixFSPathResolver: pathresolver.NewBasicResolver(fetchers.OfflineUnixfsFetcher), |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | // Dag creates new DAGService |
| 201 | func Dag(bs blockservice.BlockService) format.DAGService { |
| 202 | return merkledag.NewDAGService(bs) |
| 203 | } |
| 204 | |
| 205 | // Files loads persisted MFS root |
| 206 | func Files(strategy string) func(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService, bs blockstore.Blockstore, prov DHTProvider) (*mfs.Root, error) { |
| 207 | return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService, bs blockstore.Blockstore, prov DHTProvider) (*mfs.Root, error) { |
| 208 | pf := func(ctx context.Context, c cid.Cid) error { |
| 209 | rootDS := repo.Datastore() |
| 210 | if err := rootDS.Sync(ctx, blockstore.BlockPrefix); err != nil { |
| 211 | return err |
| 212 | } |
| 213 | if err := rootDS.Sync(ctx, filestore.FilestorePrefix); err != nil { |
| 214 | return err |
| 215 | } |
| 216 | |
| 217 | if err := rootDS.Put(ctx, FilesRootDatastoreKey, c.Bytes()); err != nil { |
| 218 | return err |
| 219 | } |
| 220 | return rootDS.Sync(ctx, FilesRootDatastoreKey) |
| 221 | } |
| 222 | |
| 223 | var nd *merkledag.ProtoNode |
| 224 | ctx := helpers.LifecycleCtx(mctx, lc) |
| 225 | val, err := repo.Datastore().Get(ctx, FilesRootDatastoreKey) |
| 226 | |
| 227 | switch { |
| 228 | case errors.Is(err, datastore.ErrNotFound): |
| 229 | nd = unixfs.EmptyDirNode() |
| 230 | err := dag.Add(ctx, nd) |
| 231 | if err != nil { |
| 232 | return nil, fmt.Errorf("failure writing filesroot to dagstore: %s", err) |
| 233 | } |
| 234 | case err == nil: |
| 235 | c, err := cid.Cast(val) |
| 236 | if err != nil { |
| 237 | return nil, err |
| 238 | } |
| 239 | |
| 240 | offlineDag := merkledag.NewDAGService(blockservice.New(bs, offline.Exchange(bs))) |
| 241 | rnd, err := offlineDag.Get(ctx, c) |
| 242 | if err != nil { |
| 243 | return nil, fmt.Errorf("error loading filesroot from dagservice: %s", err) |
| 244 | } |
| 245 | |
| 246 | pbnd, ok := rnd.(*merkledag.ProtoNode) |
| 247 | if !ok { |
| 248 | return nil, merkledag.ErrNotProtobuf |
| 249 | } |
| 250 | |
| 251 | nd = pbnd |
| 252 | default: |
| 253 | return nil, err |
| 254 | } |
| 255 | |
| 256 | // MFS (Mutable File System) provider integration: Only pass the provider |
| 257 | // to MFS when the strategy includes "mfs". MFS will call StartProviding() |
| 258 | // on every DAGService.Add() operation, which is sufficient for the "mfs" |
| 259 | // strategy - it ensures all MFS content gets announced as it's added or |
| 260 | // modified. For non-mfs strategies, we set provider to nil to avoid |
| 261 | // unnecessary providing. |
| 262 | strategyFlag := config.MustParseProvideStrategy(strategy) |
| 263 | if strategyFlag&config.ProvideStrategyMFS == 0 { |
| 264 | prov = nil |
| 265 | } |
| 266 | |
| 267 | // Get configured settings from Import config |
| 268 | cfg, err := repo.Config() |
| 269 | if err != nil { |
| 270 | return nil, fmt.Errorf("failed to get config: %w", err) |
| 271 | } |
| 272 | mfsOpts, err := cfg.Import.MFSRootOptions() |
| 273 | if err != nil { |
| 274 | return nil, fmt.Errorf("failed to build MFS options from Import config: %w", err) |
| 275 | } |
| 276 | |
| 277 | root, err := mfs.NewRoot(ctx, dag, nd, pf, prov, mfsOpts...) |
| 278 | if err != nil { |
| 279 | return nil, fmt.Errorf("failed to initialize MFS root from %s stored at %s: %w. "+ |
| 280 | "If corrupted, use 'ipfs files chroot' to reset (see --help)", nd.Cid(), FilesRootDatastoreKey, err) |
| 281 | } |
| 282 | |
| 283 | lc.Append(fx.Hook{ |
| 284 | OnStop: func(ctx context.Context) error { |
| 285 | return shutdown.CloseWithCtx(ctx, "mfs-root", root.Close) |
| 286 | }, |
| 287 | }) |
| 288 | |
| 289 | return root, err |
| 290 | } |
| 291 | } |