| 1 | package coreapi |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | |
| 8 | blockservice "github.com/ipfs/boxo/blockservice" |
| 9 | bstore "github.com/ipfs/boxo/blockstore" |
| 10 | "github.com/ipfs/boxo/files" |
| 11 | filestore "github.com/ipfs/boxo/filestore" |
| 12 | merkledag "github.com/ipfs/boxo/ipld/merkledag" |
| 13 | dagtest "github.com/ipfs/boxo/ipld/merkledag/test" |
| 14 | ft "github.com/ipfs/boxo/ipld/unixfs" |
| 15 | unixfile "github.com/ipfs/boxo/ipld/unixfs/file" |
| 16 | uio "github.com/ipfs/boxo/ipld/unixfs/io" |
| 17 | "github.com/ipfs/boxo/mfs" |
| 18 | "github.com/ipfs/boxo/path" |
| 19 | cid "github.com/ipfs/go-cid" |
| 20 | cidutil "github.com/ipfs/go-cidutil" |
| 21 | ds "github.com/ipfs/go-datastore" |
| 22 | dssync "github.com/ipfs/go-datastore/sync" |
| 23 | ipld "github.com/ipfs/go-ipld-format" |
| 24 | logging "github.com/ipfs/go-log/v2" |
| 25 | "github.com/ipfs/kubo/config" |
| 26 | coreiface "github.com/ipfs/kubo/core/coreiface" |
| 27 | options "github.com/ipfs/kubo/core/coreiface/options" |
| 28 | "github.com/ipfs/kubo/core/coreunix" |
| 29 | "github.com/ipfs/kubo/tracing" |
| 30 | "go.opentelemetry.io/otel/attribute" |
| 31 | "go.opentelemetry.io/otel/trace" |
| 32 | ) |
| 33 | |
| 34 | var log = logging.Logger("coreapi") |
| 35 | |
| 36 | type UnixfsAPI CoreAPI |
| 37 | |
| 38 | // Add builds a merkledag node from a reader, adds it to the blockstore, |
| 39 | // and returns the key representing that node. |
| 40 | func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options.UnixfsAddOption) (path.ImmutablePath, error) { |
| 41 | ctx, span := tracing.Span(ctx, "CoreAPI.UnixfsAPI", "Add") |
| 42 | defer span.End() |
| 43 | |
| 44 | settings, prefix, err := options.UnixfsAddOptions(opts...) |
| 45 | if err != nil { |
| 46 | return path.ImmutablePath{}, err |
| 47 | } |
| 48 | |
| 49 | span.SetAttributes( |
| 50 | attribute.String("chunker", settings.Chunker), |
| 51 | attribute.Int("cidversion", settings.CidVersion), |
| 52 | attribute.Bool("inline", settings.Inline), |
| 53 | attribute.Int("inlinelimit", settings.InlineLimit), |
| 54 | attribute.Bool("rawleaves", settings.RawLeaves), |
| 55 | attribute.Bool("rawleavesset", settings.RawLeavesSet), |
| 56 | attribute.Int("maxfilelinks", settings.MaxFileLinks), |
| 57 | attribute.Bool("maxfilelinksset", settings.MaxFileLinksSet), |
| 58 | attribute.Int("maxdirectorylinks", settings.MaxDirectoryLinks), |
| 59 | attribute.Bool("maxdirectorylinksset", settings.MaxDirectoryLinksSet), |
| 60 | attribute.Int("maxhamtfanout", settings.MaxHAMTFanout), |
| 61 | attribute.Bool("maxhamtfanoutset", settings.MaxHAMTFanoutSet), |
| 62 | attribute.Int("layout", int(settings.Layout)), |
| 63 | attribute.Bool("pin", settings.Pin), |
| 64 | attribute.String("pin-name", settings.PinName), |
| 65 | attribute.Bool("onlyhash", settings.OnlyHash), |
| 66 | attribute.Bool("fscache", settings.FsCache), |
| 67 | attribute.Bool("nocopy", settings.NoCopy), |
| 68 | attribute.Bool("silent", settings.Silent), |
| 69 | attribute.Bool("progress", settings.Progress), |
| 70 | ) |
| 71 | |
| 72 | cfg, err := api.repo.Config() |
| 73 | if err != nil { |
| 74 | return path.ImmutablePath{}, err |
| 75 | } |
| 76 | |
| 77 | // check if repo will exceed storage limit if added |
| 78 | // TODO: this doesn't handle the case if the hashed file is already in blocks (deduplicated) |
| 79 | // TODO: conditional GC is disabled due to it is somehow not possible to pass the size to the daemon |
| 80 | //if err := corerepo.ConditionalGC(req.Context(), n, uint64(size)); err != nil { |
| 81 | // res.SetError(err, cmds.ErrNormal) |
| 82 | // return |
| 83 | //} |
| 84 | |
| 85 | if settings.NoCopy && !(cfg.Experimental.FilestoreEnabled || cfg.Experimental.UrlstoreEnabled) { |
| 86 | return path.ImmutablePath{}, errors.New("either the filestore or the urlstore must be enabled to use nocopy, see: https://github.com/ipfs/kubo/blob/master/docs/experimental-features.md#ipfs-filestore") |
| 87 | } |
| 88 | |
| 89 | addblockstore := api.blockstore |
| 90 | if !(settings.FsCache || settings.NoCopy) { |
| 91 | addblockstore = bstore.NewGCBlockstore(api.baseBlocks, api.blockstore) |
| 92 | } |
| 93 | exch := api.exchange |
| 94 | pinning := api.pinning |
| 95 | |
| 96 | if settings.OnlyHash { |
| 97 | // setup a /dev/null pipeline to simulate adding the data |
| 98 | dstore := dssync.MutexWrap(ds.NewNullDatastore()) |
| 99 | bs := bstore.NewBlockstore(dstore, bstore.WriteThrough(true)) // we use NewNullDatastore, so ok to always WriteThrough when OnlyHash |
| 100 | addblockstore = bstore.NewGCBlockstore(bs, nil) // gclocker will never be used |
| 101 | exch = nil // exchange will never be used |
| 102 | pinning = nil // pinner will never be used |
| 103 | } |
| 104 | |
| 105 | bserv := blockservice.New(addblockstore, exch, |
| 106 | blockservice.WriteThrough(cfg.Datastore.WriteThrough.WithDefault(config.DefaultWriteThrough)), |
| 107 | ) // hash security 001 |
| 108 | |
| 109 | var dserv ipld.DAGService = merkledag.NewDAGService(bserv) |
| 110 | |
| 111 | // Per-block providing for new content is handled outside the add |
| 112 | // pipeline: |
| 113 | // |
| 114 | // - Provide.Strategy=all: every block is provided at the |
| 115 | // blockstore level via the blockstore.Provider hook |
| 116 | // (see core/node/storage.go). |
| 117 | // - Selective strategies (pinned, mfs, +unique, +entities) with |
| 118 | // --fast-provide-dag: ExecuteFastProvideDAG walks the DAG once |
| 119 | // after add completes, applying the active strategy and bloom |
| 120 | // dedup. Wiring lives in core/commands/add.go. |
| 121 | // - --fast-provide-root only (default): the root CID is announced |
| 122 | // immediately via ExecuteFastProvideRoot in the command handler. |
| 123 | // |
| 124 | // The coreapi layer therefore does not wrap the DAGService with |
| 125 | // any providing logic. |
| 126 | |
| 127 | // add a sync call to the DagService |
| 128 | // this ensures that data written to the DagService is persisted to the underlying datastore |
| 129 | // TODO: propagate the Sync function from the datastore through the blockstore, blockservice and dagservice |
| 130 | var syncDserv *syncDagService |
| 131 | if settings.OnlyHash { |
| 132 | syncDserv = &syncDagService{ |
| 133 | DAGService: dserv, |
| 134 | syncFn: func() error { return nil }, |
| 135 | } |
| 136 | } else { |
| 137 | syncDserv = &syncDagService{ |
| 138 | DAGService: dserv, |
| 139 | syncFn: func() error { |
| 140 | rds := api.repo.Datastore() |
| 141 | if err := rds.Sync(ctx, bstore.BlockPrefix); err != nil { |
| 142 | return err |
| 143 | } |
| 144 | return rds.Sync(ctx, filestore.FilestorePrefix) |
| 145 | }, |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | // Note: the dag service gets wrapped multiple times: |
| 150 | // 1. syncDagService - ensures data persistence |
| 151 | // 2. batchingDagService (in coreunix.Adder) - batches operations for efficiency |
| 152 | |
| 153 | fileAdder, err := coreunix.NewAdder(ctx, pinning, addblockstore, syncDserv) |
| 154 | if err != nil { |
| 155 | return path.ImmutablePath{}, err |
| 156 | } |
| 157 | |
| 158 | fileAdder.Chunker = settings.Chunker |
| 159 | if settings.Events != nil { |
| 160 | fileAdder.Out = settings.Events |
| 161 | fileAdder.Progress = settings.Progress |
| 162 | } |
| 163 | fileAdder.Pin = settings.Pin && !settings.OnlyHash |
| 164 | if settings.Pin { |
| 165 | fileAdder.PinName = settings.PinName |
| 166 | } |
| 167 | fileAdder.Silent = settings.Silent |
| 168 | fileAdder.RawLeaves = settings.RawLeaves |
| 169 | if settings.MaxFileLinksSet { |
| 170 | fileAdder.MaxLinks = settings.MaxFileLinks |
| 171 | } |
| 172 | if settings.MaxDirectoryLinksSet { |
| 173 | fileAdder.MaxDirectoryLinks = settings.MaxDirectoryLinks |
| 174 | } |
| 175 | |
| 176 | if settings.MaxHAMTFanoutSet { |
| 177 | fileAdder.MaxHAMTFanout = settings.MaxHAMTFanout |
| 178 | } |
| 179 | if settings.SizeEstimationModeSet { |
| 180 | fileAdder.SizeEstimationMode = settings.SizeEstimationMode |
| 181 | } |
| 182 | fileAdder.NoCopy = settings.NoCopy |
| 183 | fileAdder.CidBuilder = prefix |
| 184 | fileAdder.PreserveMode = settings.PreserveMode |
| 185 | fileAdder.PreserveMtime = settings.PreserveMtime |
| 186 | fileAdder.FileMode = settings.Mode |
| 187 | fileAdder.FileMtime = settings.Mtime |
| 188 | if settings.IncludeEmptyDirsSet { |
| 189 | fileAdder.IncludeEmptyDirs = settings.IncludeEmptyDirs |
| 190 | } |
| 191 | |
| 192 | switch settings.Layout { |
| 193 | case options.BalancedLayout: |
| 194 | // Default |
| 195 | case options.TrickleLayout: |
| 196 | fileAdder.Trickle = true |
| 197 | default: |
| 198 | return path.ImmutablePath{}, fmt.Errorf("unknown layout: %d", settings.Layout) |
| 199 | } |
| 200 | |
| 201 | if settings.Inline { |
| 202 | fileAdder.CidBuilder = cidutil.InlineBuilder{ |
| 203 | Builder: fileAdder.CidBuilder, |
| 204 | Limit: settings.InlineLimit, |
| 205 | } |
| 206 | } |
| 207 | |
| 208 | if settings.OnlyHash { |
| 209 | md := dagtest.Mock() |
| 210 | emptyDirNode := ft.EmptyDirNode() |
| 211 | // Use the same prefix for the "empty" MFS root as for the file adder. |
| 212 | err := emptyDirNode.SetCidBuilder(fileAdder.CidBuilder) |
| 213 | if err != nil { |
| 214 | return path.ImmutablePath{}, err |
| 215 | } |
| 216 | // MFS root for OnlyHash mode: provider is nil since we're not storing/providing anything |
| 217 | mr, err := mfs.NewRoot(ctx, md, emptyDirNode, nil, nil) |
| 218 | if err != nil { |
| 219 | return path.ImmutablePath{}, err |
| 220 | } |
| 221 | |
| 222 | fileAdder.SetMfsRoot(mr) |
| 223 | } |
| 224 | |
| 225 | nd, err := fileAdder.AddAllAndPin(ctx, files) |
| 226 | if err != nil { |
| 227 | return path.ImmutablePath{}, err |
| 228 | } |
| 229 | |
| 230 | return path.FromCid(nd.Cid()), nil |
| 231 | } |
| 232 | |
| 233 | func (api *UnixfsAPI) Get(ctx context.Context, p path.Path) (files.Node, error) { |
| 234 | ctx, span := tracing.Span(ctx, "CoreAPI.UnixfsAPI", "Get", trace.WithAttributes(attribute.String("path", p.String()))) |
| 235 | defer span.End() |
| 236 | |
| 237 | ses := api.core().getSession(ctx) |
| 238 | |
| 239 | nd, err := ses.ResolveNode(ctx, p) |
| 240 | if err != nil { |
| 241 | return nil, err |
| 242 | } |
| 243 | |
| 244 | return unixfile.NewUnixfsFile(ctx, ses.dag, nd) |
| 245 | } |
| 246 | |
| 247 | // Ls returns the contents of an IPFS or IPNS object(s) at path p, with the format: |
| 248 | // `<link base58 hash> <link size in bytes> <link name>` |
| 249 | func (api *UnixfsAPI) Ls(ctx context.Context, p path.Path, out chan<- coreiface.DirEntry, opts ...options.UnixfsLsOption) error { |
| 250 | ctx, span := tracing.Span(ctx, "CoreAPI.UnixfsAPI", "Ls", trace.WithAttributes(attribute.String("path", p.String()))) |
| 251 | defer span.End() |
| 252 | |
| 253 | defer close(out) |
| 254 | |
| 255 | settings, err := options.UnixfsLsOptions(opts...) |
| 256 | if err != nil { |
| 257 | return err |
| 258 | } |
| 259 | |
| 260 | span.SetAttributes(attribute.Bool("resolvechildren", settings.ResolveChildren)) |
| 261 | |
| 262 | ses := api.core().getSession(ctx) |
| 263 | uses := (*UnixfsAPI)(ses) |
| 264 | |
| 265 | dagnode, err := ses.ResolveNode(ctx, p) |
| 266 | if err != nil { |
| 267 | return err |
| 268 | } |
| 269 | |
| 270 | dir, err := uio.NewDirectoryFromNode(ses.dag, dagnode) |
| 271 | if err != nil { |
| 272 | if errors.Is(err, uio.ErrNotADir) { |
| 273 | return uses.lsFromLinks(ctx, dagnode.Links(), settings, out) |
| 274 | } |
| 275 | return err |
| 276 | } |
| 277 | |
| 278 | return uses.lsFromDirLinks(ctx, dir, settings, out) |
| 279 | } |
| 280 | |
| 281 | func (api *UnixfsAPI) processLink(ctx context.Context, linkres ft.LinkResult, settings *options.UnixfsLsSettings) (coreiface.DirEntry, error) { |
| 282 | ctx, span := tracing.Span(ctx, "CoreAPI.UnixfsAPI", "ProcessLink") |
| 283 | defer span.End() |
| 284 | if linkres.Link != nil { |
| 285 | span.SetAttributes(attribute.String("linkname", linkres.Link.Name), attribute.String("cid", linkres.Link.Cid.String())) |
| 286 | } |
| 287 | |
| 288 | if linkres.Err != nil { |
| 289 | return coreiface.DirEntry{}, linkres.Err |
| 290 | } |
| 291 | |
| 292 | lnk := coreiface.DirEntry{ |
| 293 | Name: linkres.Link.Name, |
| 294 | Cid: linkres.Link.Cid, |
| 295 | } |
| 296 | |
| 297 | switch lnk.Cid.Type() { |
| 298 | case cid.Raw: |
| 299 | // No need to check with raw leaves |
| 300 | lnk.Type = coreiface.TFile |
| 301 | lnk.Size = linkres.Link.Size |
| 302 | case cid.DagProtobuf: |
| 303 | if settings.ResolveChildren { |
| 304 | linkNode, err := linkres.Link.GetNode(ctx, api.dag) |
| 305 | if err != nil { |
| 306 | return coreiface.DirEntry{}, err |
| 307 | } |
| 308 | |
| 309 | if pn, ok := linkNode.(*merkledag.ProtoNode); ok { |
| 310 | d, err := ft.FSNodeFromBytes(pn.Data()) |
| 311 | if err != nil { |
| 312 | return coreiface.DirEntry{}, err |
| 313 | } |
| 314 | switch d.Type() { |
| 315 | case ft.TFile, ft.TRaw: |
| 316 | lnk.Type = coreiface.TFile |
| 317 | case ft.THAMTShard, ft.TDirectory, ft.TMetadata: |
| 318 | lnk.Type = coreiface.TDirectory |
| 319 | case ft.TSymlink: |
| 320 | lnk.Type = coreiface.TSymlink |
| 321 | lnk.Target = string(d.Data()) |
| 322 | } |
| 323 | if !settings.UseCumulativeSize { |
| 324 | lnk.Size = d.FileSize() |
| 325 | } |
| 326 | lnk.Mode = d.Mode() |
| 327 | lnk.ModTime = d.ModTime() |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | if settings.UseCumulativeSize { |
| 332 | lnk.Size = linkres.Link.Size |
| 333 | } |
| 334 | } |
| 335 | |
| 336 | return lnk, nil |
| 337 | } |
| 338 | |
| 339 | func (api *UnixfsAPI) lsFromDirLinks(ctx context.Context, dir uio.Directory, settings *options.UnixfsLsSettings, out chan<- coreiface.DirEntry) error { |
| 340 | for l := range dir.EnumLinksAsync(ctx) { |
| 341 | dirEnt, err := api.processLink(ctx, l, settings) // TODO: perf: processing can be done in background and in parallel |
| 342 | if err != nil { |
| 343 | return err |
| 344 | } |
| 345 | select { |
| 346 | case out <- dirEnt: |
| 347 | case <-ctx.Done(): |
| 348 | return nil |
| 349 | } |
| 350 | } |
| 351 | return nil |
| 352 | } |
| 353 | |
| 354 | func (api *UnixfsAPI) lsFromLinks(ctx context.Context, ndlinks []*ipld.Link, settings *options.UnixfsLsSettings, out chan<- coreiface.DirEntry) error { |
| 355 | // Create links channel large enough to not block when writing to out is slower. |
| 356 | links := make(chan coreiface.DirEntry, len(ndlinks)) |
| 357 | errs := make(chan error, 1) |
| 358 | go func() { |
| 359 | defer close(links) |
| 360 | defer close(errs) |
| 361 | for _, l := range ndlinks { |
| 362 | lr := ft.LinkResult{Link: &ipld.Link{Name: l.Name, Size: l.Size, Cid: l.Cid}} |
| 363 | lnk, err := api.processLink(ctx, lr, settings) // TODO: can be parallel if settings.Async |
| 364 | if err != nil { |
| 365 | errs <- err |
| 366 | return |
| 367 | } |
| 368 | select { |
| 369 | case links <- lnk: |
| 370 | case <-ctx.Done(): |
| 371 | return |
| 372 | } |
| 373 | } |
| 374 | }() |
| 375 | |
| 376 | for lnk := range links { |
| 377 | out <- lnk |
| 378 | } |
| 379 | return <-errs |
| 380 | } |
| 381 | |
| 382 | func (api *UnixfsAPI) core() *CoreAPI { |
| 383 | return (*CoreAPI)(api) |
| 384 | } |
| 385 | |
| 386 | // syncDagService is used by the Adder to ensure blocks get persisted to the underlying datastore |
| 387 | type syncDagService struct { |
| 388 | ipld.DAGService |
| 389 | syncFn func() error |
| 390 | } |
| 391 | |
| 392 | func (s *syncDagService) Sync() error { |
| 393 | return s.syncFn() |
| 394 | } |