| 1 | package dagcmd |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | |
| 8 | "github.com/ipfs/boxo/files" |
| 9 | blocks "github.com/ipfs/go-block-format" |
| 10 | cid "github.com/ipfs/go-cid" |
| 11 | cmds "github.com/ipfs/go-ipfs-cmds" |
| 12 | ipld "github.com/ipfs/go-ipld-format" |
| 13 | ipldlegacy "github.com/ipfs/go-ipld-legacy" |
| 14 | logging "github.com/ipfs/go-log/v2" |
| 15 | "github.com/ipfs/kubo/config" |
| 16 | "github.com/ipfs/kubo/core/coreiface/options" |
| 17 | gocarv2 "github.com/ipld/go-car/v2" |
| 18 | |
| 19 | "github.com/ipfs/kubo/core/commands/cmdenv" |
| 20 | "github.com/ipfs/kubo/core/commands/cmdutils" |
| 21 | ) |
| 22 | |
| 23 | var log = logging.Logger("core/commands") |
| 24 | |
| 25 | func dagImport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 26 | node, err := cmdenv.GetNode(env) |
| 27 | if err != nil { |
| 28 | return err |
| 29 | } |
| 30 | |
| 31 | cfg, err := node.Repo.Config() |
| 32 | if err != nil { |
| 33 | return err |
| 34 | } |
| 35 | |
| 36 | api, err := cmdenv.GetApi(env, req) |
| 37 | if err != nil { |
| 38 | return err |
| 39 | } |
| 40 | |
| 41 | blockDecoder := ipldlegacy.NewDecoder() |
| 42 | |
| 43 | // on import ensure we do not reach out to the network for any reason |
| 44 | // if a pin based on what is imported + what is in the blockstore |
| 45 | // isn't possible: tough luck |
| 46 | api, err = api.WithOptions(options.Api.Offline(true)) |
| 47 | if err != nil { |
| 48 | return err |
| 49 | } |
| 50 | |
| 51 | pinRootsVal, pinRootsSet := req.Options[pinRootsOptionName].(bool) |
| 52 | localOnly, _ := req.Options[localOnlyOptionName].(bool) |
| 53 | |
| 54 | // --pin-roots defaults to true; the default is applied here (not via |
| 55 | // .WithDefault) so we can tell apart "user explicitly passed true" from |
| 56 | // "no value provided". |
| 57 | doPinRoots := true |
| 58 | if pinRootsSet { |
| 59 | doPinRoots = pinRootsVal |
| 60 | } |
| 61 | |
| 62 | if localOnly { |
| 63 | if pinRootsSet && pinRootsVal { |
| 64 | return fmt.Errorf("--%s implies --%s=false and cannot be combined with --%s=true; please drop one of them", localOnlyOptionName, pinRootsOptionName, pinRootsOptionName) |
| 65 | } |
| 66 | // --local-only implies --pin-roots=false: a partial CAR has no full DAG to pin. |
| 67 | doPinRoots = false |
| 68 | } |
| 69 | fastProvideRoot, fastProvideRootSet := req.Options[fastProvideRootOptionName].(bool) |
| 70 | fastProvideDAG, fastProvideDAGSet := req.Options[fastProvideDAGOptionName].(bool) |
| 71 | fastProvideWait, fastProvideWaitSet := req.Options[fastProvideWaitOptionName].(bool) |
| 72 | |
| 73 | fastProvideRoot = config.ResolveBoolFromConfig(fastProvideRoot, fastProvideRootSet, cfg.Import.FastProvideRoot, config.DefaultFastProvideRoot) |
| 74 | fastProvideDAG = config.ResolveBoolFromConfig(fastProvideDAG, fastProvideDAGSet, cfg.Import.FastProvideDAG, config.DefaultFastProvideDAG) |
| 75 | fastProvideWait = config.ResolveBoolFromConfig(fastProvideWait, fastProvideWaitSet, cfg.Import.FastProvideWait, config.DefaultFastProvideWait) |
| 76 | |
| 77 | // grab a pinlock ( which doubles as a GC lock ) so that regardless of the |
| 78 | // size of the streamed-in cars nothing will disappear on us before we had |
| 79 | // a chance to roots that may show up at the very end |
| 80 | // This is especially important for use cases like dagger: |
| 81 | // ipfs dag import $( ... | ipfs-dagger --stdout=carfifos ) |
| 82 | // |
| 83 | if doPinRoots { |
| 84 | unlocker := node.Blockstore.PinLock(req.Context) |
| 85 | defer unlocker.Unlock(req.Context) |
| 86 | } |
| 87 | |
| 88 | // this is *not* a transaction |
| 89 | // it is simply a way to relieve pressure on the blockstore |
| 90 | // similar to pinner.Pin/pinner.Flush |
| 91 | batch := ipld.NewBatch(req.Context, api.Dag(), |
| 92 | // Default: 128. Means 128 file descriptors needed in flatfs |
| 93 | ipld.MaxNodesBatchOption(int(cfg.Import.BatchMaxNodes.WithDefault(config.DefaultBatchMaxNodes))), |
| 94 | // Default 100MiB. When setting block size to 1MiB, we can add |
| 95 | // ~100 nodes maximum. With default 256KiB block-size, we will |
| 96 | // hit the max nodes limit at 32MiB.p |
| 97 | ipld.MaxSizeBatchOption(int(cfg.Import.BatchMaxSize.WithDefault(config.DefaultBatchMaxSize))), |
| 98 | ) |
| 99 | |
| 100 | roots := cid.NewSet() |
| 101 | var blockCount, blockBytesCount uint64 |
| 102 | |
| 103 | // remember last valid block and provide a meaningful error message |
| 104 | // when a truncated/mangled CAR is being imported |
| 105 | importError := func(previous blocks.Block, current blocks.Block, err error) error { |
| 106 | if current != nil { |
| 107 | return fmt.Errorf("import failed at block %q: %w", current.Cid(), err) |
| 108 | } |
| 109 | if previous != nil { |
| 110 | return fmt.Errorf("import failed after block %q: %w", previous.Cid(), err) |
| 111 | } |
| 112 | return fmt.Errorf("import failed: %w", err) |
| 113 | } |
| 114 | |
| 115 | it := req.Files.Entries() |
| 116 | for it.Next() { |
| 117 | file := files.FileFromEntry(it) |
| 118 | if file == nil { |
| 119 | return errors.New("expected a file handle") |
| 120 | } |
| 121 | |
| 122 | // import blocks |
| 123 | err = func() error { |
| 124 | // wrap a defer-closer-scope |
| 125 | // |
| 126 | // every single file in it() is already open before we start |
| 127 | // just close here sooner rather than later for neatness |
| 128 | // and to surface potential errors writing on closed fifos |
| 129 | // this won't/can't help with not running out of handles |
| 130 | defer file.Close() |
| 131 | |
| 132 | var previous blocks.Block |
| 133 | |
| 134 | // Wrap the file to hide the io.Seeker interface. |
| 135 | // Over the HTTP API the underlying reader is a multipart stream |
| 136 | // that cannot seek, but boxo's ReaderFile advertises io.Seeker |
| 137 | // anyway and returns ErrNotSupported at runtime. Hiding the |
| 138 | // interface lets go-car fall back to sequential (forward-only) |
| 139 | // reading, which is all that CARv2 streaming needs. |
| 140 | // See https://github.com/ipfs/kubo/issues/9361 |
| 141 | car, err := gocarv2.NewBlockReader(struct { |
| 142 | io.Reader |
| 143 | io.Closer |
| 144 | }{file, file}) |
| 145 | if err != nil { |
| 146 | return err |
| 147 | } |
| 148 | |
| 149 | for _, c := range car.Roots { |
| 150 | roots.Add(c) |
| 151 | } |
| 152 | |
| 153 | for { |
| 154 | block, err := car.Next() |
| 155 | if err != nil && err != io.EOF { |
| 156 | return importError(previous, block, err) |
| 157 | } else if block == nil { |
| 158 | break |
| 159 | } |
| 160 | if err := cmdutils.CheckBlockSize(req, uint64(len(block.RawData()))); err != nil { |
| 161 | return importError(previous, block, err) |
| 162 | } |
| 163 | |
| 164 | // the double-decode is suboptimal, but we need it for batching |
| 165 | nd, err := blockDecoder.DecodeNode(req.Context, block) |
| 166 | if err != nil { |
| 167 | return importError(previous, block, err) |
| 168 | } |
| 169 | |
| 170 | if err := batch.Add(req.Context, nd); err != nil { |
| 171 | return importError(previous, block, err) |
| 172 | } |
| 173 | blockCount++ |
| 174 | blockBytesCount += uint64(len(block.RawData())) |
| 175 | previous = block |
| 176 | } |
| 177 | return nil |
| 178 | }() |
| 179 | if err != nil { |
| 180 | return err |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | if err := batch.Commit(); err != nil { |
| 185 | return err |
| 186 | } |
| 187 | |
| 188 | // It is not guaranteed that a root in a header is actually present in the same ( or any ) |
| 189 | // .car file. This is the case in version 1, and ideally in further versions too. |
| 190 | // Accumulate any root CID seen in a header, and supplement its actual node if/when encountered |
| 191 | // We will attempt a pin *only* at the end in case all car files were well-formed. |
| 192 | |
| 193 | // opportunistic pinning: try whatever sticks |
| 194 | if doPinRoots { |
| 195 | err = roots.ForEach(func(c cid.Cid) error { |
| 196 | ret := RootMeta{Cid: c} |
| 197 | |
| 198 | // This will trigger a full read of the DAG in the pinner, to make sure we have all blocks. |
| 199 | // Ideally we would do colloring of the pinning state while importing the blocks |
| 200 | // and ensure the gray bucket is empty at the end (or use the network to download missing blocks). |
| 201 | if block, err := node.Blockstore.Get(req.Context, c); err != nil { |
| 202 | ret.PinErrorMsg = err.Error() |
| 203 | } else if nd, err := blockDecoder.DecodeNode(req.Context, block); err != nil { |
| 204 | ret.PinErrorMsg = err.Error() |
| 205 | } else if err := node.Pinning.Pin(req.Context, nd, true, ""); err != nil { |
| 206 | ret.PinErrorMsg = err.Error() |
| 207 | } else if err := node.Pinning.Flush(req.Context); err != nil { |
| 208 | ret.PinErrorMsg = err.Error() |
| 209 | } |
| 210 | |
| 211 | return res.Emit(&CarImportOutput{Root: &ret}) |
| 212 | }) |
| 213 | if err != nil { |
| 214 | return err |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | stats, _ := req.Options[statsOptionName].(bool) |
| 219 | if stats { |
| 220 | err = res.Emit(&CarImportOutput{ |
| 221 | Stats: &CarImportStats{ |
| 222 | BlockCount: blockCount, |
| 223 | BlockBytesCount: blockBytesCount, |
| 224 | }, |
| 225 | }) |
| 226 | if err != nil { |
| 227 | return err |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | // Provide imported content for faster discovery. |
| 232 | // DAG walk supersedes root-only (root is included in the walk). |
| 233 | if fastProvideDAG { |
| 234 | var rootCIDs []cid.Cid |
| 235 | _ = roots.ForEach(func(c cid.Cid) error { |
| 236 | rootCIDs = append(rootCIDs, c) |
| 237 | return nil |
| 238 | }) |
| 239 | cmdenv.ExecuteFastProvideDAG( |
| 240 | req.Context, |
| 241 | node.Context(), |
| 242 | rootCIDs, |
| 243 | node.ProvidingStrategy, |
| 244 | node.Blockstore, |
| 245 | node.Provider, |
| 246 | fastProvideWait, |
| 247 | uint(cfg.Provide.BloomFPRate.WithDefault(config.DefaultProvideBloomFPRate)), |
| 248 | 0, // block count unknown; bloom chain auto-grows |
| 249 | ) |
| 250 | } else if fastProvideRoot { |
| 251 | err = roots.ForEach(func(c cid.Cid) error { |
| 252 | return cmdenv.ExecuteFastProvideRoot(req.Context, node, cfg, c, fastProvideWait, doPinRoots, doPinRoots, false) |
| 253 | }) |
| 254 | if err != nil { |
| 255 | return err |
| 256 | } |
| 257 | } else { |
| 258 | log.Debugw("fast-provide-root: skipped", "reason", "disabled by flag or config") |
| 259 | } |
| 260 | |
| 261 | return nil |
| 262 | } |