@cryptotaxi247 / kubo / commits / 6e2b16677

split core/commands/dag into individual files for different subcommands

Will Scott committed Mar 10, 2021 at 12:14 UTC 6e2b1667733c74ca25780d29829474a5cda15baa
7 files changed +606 -518
core/commands/dag/dag.go
+17 -518
@@ -1,37 +1,18 @@
1 package dagcmd
2
3 import (
4 - "errors"
4 "fmt"
5 "io"
7 - "math"
8 - "os"
9 - "strings"
10 - "time"
6
7 "github.com/ipfs/go-ipfs/core/commands/cmdenv"
13 - "github.com/ipfs/go-ipfs/core/commands/e"
14 - "github.com/ipfs/go-ipfs/core/coredag"
15 - iface "github.com/ipfs/interface-go-ipfs-core"
8
9 cid "github.com/ipfs/go-cid"
10 cidenc "github.com/ipfs/go-cidutil/cidenc"
11 cmds "github.com/ipfs/go-ipfs-cmds"
20 - files "github.com/ipfs/go-ipfs-files"
21 - ipld "github.com/ipfs/go-ipld-format"
22 - mdag "github.com/ipfs/go-merkledag"
23 - traverse "github.com/ipfs/go-merkledag/traverse"
12 ipfspath "github.com/ipfs/go-path"
25 - "github.com/ipfs/interface-go-ipfs-core/options"
26 - path "github.com/ipfs/interface-go-ipfs-core/path"
27 - mh "github.com/multiformats/go-multihash"
28 -
29 - gocar "github.com/ipld/go-car"
13 //gipfree "github.com/ipld/go-ipld-prime/impl/free"
14 //gipselector "github.com/ipld/go-ipld-prime/traversal/selector"
15 //gipselectorbuilder "github.com/ipld/go-ipld-prime/traversal/selector/builder"
33 -
34 - "github.com/cheggaaa/pb"
16 )
17
18 const (
@@ -40,6 +21,7 @@ const (
21 pinRootsOptionName = "pin-roots"
22 )
23
24 +// DagCmd provides a subset of commands for interacting with ipld dag objects
25 var DagCmd = &cmds.Command{
26 Helptext: cmds.HelpText{
27 Tagline: "Interact with ipld dag objects.",
@@ -75,11 +57,14 @@ type ResolveOutput struct {
57 type CarImportOutput struct {
58 Root RootMeta
59 }
60 +
61 +// RootMeta is the metadata for a root pinning response
62 type RootMeta struct {
63 Cid cid.Cid
64 PinErrorMsg string
65 }
66
67 +// DagPutCmd is a command for adding a dag node
68 var DagPutCmd = &cmds.Command{
69 Helptext: cmds.HelpText{
70 Tagline: "Add a dag node to ipfs.",
@@ -97,71 +82,7 @@ into an object of the specified format.
82 cmds.BoolOption("pin", "Pin this object when adding."),
83 cmds.StringOption("hash", "Hash function to use").WithDefault(""),
84 },
100 - Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
101 - api, err := cmdenv.GetApi(env, req)
102 - if err != nil {
103 - return err
104 - }
105 -
106 - ienc, _ := req.Options["input-enc"].(string)
107 - format, _ := req.Options["format"].(string)
108 - hash, _ := req.Options["hash"].(string)
109 - dopin, _ := req.Options["pin"].(bool)
110 -
111 - // mhType tells inputParser which hash should be used. MaxUint64 means 'use
112 - // default hash' (sha256 for cbor, sha1 for git..)
113 - mhType := uint64(math.MaxUint64)
114 -
115 - if hash != "" {
116 - var ok bool
117 - mhType, ok = mh.Names[hash]
118 - if !ok {
119 - return fmt.Errorf("%s in not a valid multihash name", hash)
120 - }
121 - }
122 -
123 - var adder ipld.NodeAdder = api.Dag()
124 - if dopin {
125 - adder = api.Dag().Pinning()
126 - }
127 - b := ipld.NewBatch(req.Context, adder)
128 -
129 - it := req.Files.Entries()
130 - for it.Next() {
131 - file := files.FileFromEntry(it)
132 - if file == nil {
133 - return fmt.Errorf("expected a regular file")
134 - }
135 - nds, err := coredag.ParseInputs(ienc, format, file, mhType, -1)
136 - if err != nil {
137 - return err
138 - }
139 - if len(nds) == 0 {
140 - return fmt.Errorf("no node returned from ParseInputs")
141 - }
142 -
143 - for _, nd := range nds {
144 - err := b.Add(req.Context, nd)
145 - if err != nil {
146 - return err
147 - }
148 - }
149 -
150 - cid := nds[0].Cid()
151 - if err := res.Emit(&OutputObject{Cid: cid}); err != nil {
152 - return err
153 - }
154 - }
155 - if it.Err() != nil {
156 - return it.Err()
157 - }
158 -
159 - if err := b.Commit(); err != nil {
160 - return err
161 - }
162 -
163 - return nil
164 - },
85 + Run: dagPut,
86 Type: OutputObject{},
87 Encoders: cmds.EncoderMap{
88 cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *OutputObject) error {
@@ -175,6 +96,7 @@ into an object of the specified format.
96 },
97 }
98
99 +// DagGetCmd is a command for getting a dag node from IPFS
100 var DagGetCmd = &cmds.Command{
101 Helptext: cmds.HelpText{
102 Tagline: "Get a dag node from ipfs.",
@@ -186,33 +108,7 @@ format.
108 Arguments: []cmds.Argument{
109 cmds.StringArg("ref", true, false, "The object to get").EnableStdin(),
110 },
189 - Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
190 - api, err := cmdenv.GetApi(env, req)
191 - if err != nil {
192 - return err
193 - }
194 -
195 - rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
196 - if err != nil {
197 - return err
198 - }
199 -
200 - obj, err := api.Dag().Get(req.Context, rp.Cid())
201 - if err != nil {
202 - return err
203 - }
204 -
205 - var out interface{} = obj
206 - if len(rp.Remainder()) > 0 {
207 - rem := strings.Split(rp.Remainder(), "/")
208 - final, _, err := obj.Resolve(rem)
209 - if err != nil {
210 - return err
211 - }
212 - out = final
213 - }
214 - return cmds.EmitOnce(res, &out)
215 - },
111 + Run: dagGet,
112 }
113
114 // DagResolveCmd returns address of highest block within a path and a path remainder
@@ -226,22 +122,7 @@ var DagResolveCmd = &cmds.Command{
122 Arguments: []cmds.Argument{
123 cmds.StringArg("ref", true, false, "The path to resolve").EnableStdin(),
124 },
229 - Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
230 - api, err := cmdenv.GetApi(env, req)
231 - if err != nil {
232 - return err
233 - }
234 -
235 - rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
236 - if err != nil {
237 - return err
238 - }
239 -
240 - return cmds.EmitOnce(res, &ResolveOutput{
241 - Cid: rp.Cid(),
242 - RemPath: rp.Remainder(),
243 - })
244 - },
125 + Run: dagResolve,
126 Encoders: cmds.EncoderMap{
127 cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *ResolveOutput) error {
128 var (
@@ -280,6 +161,7 @@ type importResult struct {
161 err error
162 }
163
164 +// DagImportCmd is a command for importing a car to ipfs
165 var DagImportCmd = &cmds.Command{
166 Helptext: cmds.HelpText{
167 Tagline: "Import the contents of .car files",
@@ -312,107 +194,7 @@ Maximum supported CAR version: 1
194 cmds.BoolOption(pinRootsOptionName, "Pin optional roots listed in the .car headers after importing.").WithDefault(true),
195 },
196 Type: CarImportOutput{},
315 - Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
316 -
317 - node, err := cmdenv.GetNode(env)
318 - if err != nil {
319 - return err
320 - }
321 -
322 - api, err := cmdenv.GetApi(env, req)
323 - if err != nil {
324 - return err
325 - }
326 -
327 - // on import ensure we do not reach out to the network for any reason
328 - // if a pin based on what is imported + what is in the blockstore
329 - // isn't possible: tough luck
330 - api, err = api.WithOptions(options.Api.Offline(true))
331 - if err != nil {
332 - return err
333 - }
334 -
335 - // grab a pinlock ( which doubles as a GC lock ) so that regardless of the
336 - // size of the streamed-in cars nothing will disappear on us before we had
337 - // a chance to roots that may show up at the very end
338 - // This is especially important for use cases like dagger:
339 - // ipfs dag import $( ... | ipfs-dagger --stdout=carfifos )
340 - //
341 - unlocker := node.Blockstore.PinLock()
342 - defer unlocker.Unlock()
343 -
344 - doPinRoots, _ := req.Options[pinRootsOptionName].(bool)
345 -
346 - retCh := make(chan importResult, 1)
347 - go importWorker(req, res, api, retCh)
348 -
349 - done := <-retCh
350 - if done.err != nil {
351 - return done.err
352 - }
353 -
354 - // It is not guaranteed that a root in a header is actually present in the same ( or any )
355 - // .car file. This is the case in version 1, and ideally in further versions too
356 - // Accumulate any root CID seen in a header, and supplement its actual node if/when encountered
357 - // We will attempt a pin *only* at the end in case all car files were well formed
358 - //
359 - // The boolean value indicates whether we have encountered the root within the car file's
360 - roots := done.roots
361 -
362 - // opportunistic pinning: try whatever sticks
363 - if doPinRoots {
364 -
365 - var failedPins int
366 - for c := range roots {
367 -
368 - // We need to re-retrieve a block, convert it to ipld, and feed it
369 - // to the Pinning interface, sigh...
370 - //
371 - // If we didn't have the problem of inability to take multiple pinlocks,
372 - // we could use the api directly like so (though internally it does the same):
373 - //
374 - // // not ideal, but the pinning api takes only paths :(
375 - // rp := path.NewResolvedPath(
376 - // ipfspath.FromCid(c),
377 - // c,
378 - // c,
379 - // "",
380 - // )
381 - //
382 - // if err := api.Pin().Add(req.Context, rp, options.Pin.Recursive(true)); err != nil {
383 -
384 - ret := RootMeta{Cid: c}
385 -
386 - if block, err := node.Blockstore.Get(c); err != nil {
387 - ret.PinErrorMsg = err.Error()
388 - } else if nd, err := ipld.Decode(block); err != nil {
389 - ret.PinErrorMsg = err.Error()
390 - } else if err := node.Pinning.Pin(req.Context, nd, true); err != nil {
391 - ret.PinErrorMsg = err.Error()
392 - } else if err := node.Pinning.Flush(req.Context); err != nil {
393 - ret.PinErrorMsg = err.Error()
394 - }
395 -
396 - if ret.PinErrorMsg != "" {
397 - failedPins++
398 - }
399 -
400 - if err := res.Emit(&CarImportOutput{Root: ret}); err != nil {
401 - return err
402 - }
403 - }
404 -
405 - if failedPins > 0 {
406 - return fmt.Errorf(
407 - "unable to pin all roots: %d out of %d failed",
408 - failedPins,
409 - len(roots),
410 - )
411 - }
412 - }
413 -
414 - return nil
415 - },
197 + Run: dagImport,
198 Encoders: cmds.EncoderMap{
199 cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, event *CarImportOutput) error {
200
@@ -443,88 +225,7 @@ Maximum supported CAR version: 1
225 },
226 }
227
446 -func importWorker(req *cmds.Request, re cmds.ResponseEmitter, api iface.CoreAPI, ret chan importResult) {
447 -
448 - // this is *not* a transaction
449 - // it is simply a way to relieve pressure on the blockstore
450 - // similar to pinner.Pin/pinner.Flush
451 - batch := ipld.NewBatch(req.Context, api.Dag())
452 -
453 - roots := make(map[cid.Cid]struct{})
454 -
455 - it := req.Files.Entries()
456 - for it.Next() {
457 -
458 - file := files.FileFromEntry(it)
459 - if file == nil {
460 - ret <- importResult{err: errors.New("expected a file handle")}
461 - return
462 - }
463 -
464 - // wrap a defer-closer-scope
465 - //
466 - // every single file in it() is already open before we start
467 - // just close here sooner rather than later for neatness
468 - // and to surface potential errors writing on closed fifos
469 - // this won't/can't help with not running out of handles
470 - err := func() error {
471 - defer file.Close()
472 -
473 - car, err := gocar.NewCarReader(file)
474 - if err != nil {
475 - return err
476 - }
477 -
478 - // Be explicit here, until the spec is finished
479 - if car.Header.Version != 1 {
480 - return errors.New("only car files version 1 supported at present")
481 - }
482 -
483 - for _, c := range car.Header.Roots {
484 - roots[c] = struct{}{}
485 - }
486 -
487 - for {
488 - block, err := car.Next()
489 - if err != nil && err != io.EOF {
490 - return err
491 - } else if block == nil {
492 - break
493 - }
494 -
495 - // the double-decode is suboptimal, but we need it for batching
496 - nd, err := ipld.Decode(block)
497 - if err != nil {
498 - return err
499 - }
500 -
501 - if err := batch.Add(req.Context, nd); err != nil {
502 - return err
503 - }
504 - }
505 -
506 - return nil
507 - }()
508 -
509 - if err != nil {
510 - ret <- importResult{err: err}
511 - return
512 - }
513 - }
514 -
515 - if err := it.Err(); err != nil {
516 - ret <- importResult{err: err}
517 - return
518 - }
519 -
520 - if err := batch.Commit(); err != nil {
521 - ret <- importResult{err: err}
522 - return
523 - }
524 -
525 - ret <- importResult{roots: roots}
526 -}
527 -
228 +// DagExportCmd is a command for exporting an ipfs dag to a car
229 var DagExportCmd = &cmds.Command{
230 Helptext: cmds.HelpText{
231 Tagline: "Streams the selected DAG as a .car stream on stdout.",
@@ -540,145 +241,13 @@ The output of blocks happens in strict DAG-traversal, first-seen, order.
241 Options: []cmds.Option{
242 cmds.BoolOption(progressOptionName, "p", "Display progress on CLI. Defaults to true when STDERR is a TTY."),
243 },
543 - Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
544 -
545 - c, err := cid.Decode(req.Arguments[0])
546 - if err != nil {
547 - return fmt.Errorf(
548 - "unable to parse root specification (currently only bare CIDs are supported): %s",
549 - err,
550 - )
551 - }
552 -
553 - api, err := cmdenv.GetApi(env, req)
554 - if err != nil {
555 - return err
556 - }
557 -
558 - // Code disabled until descent-issue in go-ipld-prime is fixed
559 - // https://github.com/ribasushi/gip-muddle-up
560 - //
561 - // sb := gipselectorbuilder.NewSelectorSpecBuilder(gipfree.NodeBuilder())
562 - // car := gocar.NewSelectiveCar(
563 - // req.Context,
564 - // <needs to be fixed to take format.NodeGetter as well>,
565 - // []gocar.Dag{gocar.Dag{
566 - // Root: c,
567 - // Selector: sb.ExploreRecursive(
568 - // gipselector.RecursionLimitNone(),
569 - // sb.ExploreAll(sb.ExploreRecursiveEdge()),
570 - // ).Node(),
571 - // }},
572 - // )
573 - // ...
574 - // if err := car.Write(pipeW); err != nil {}
575 -
576 - pipeR, pipeW := io.Pipe()
577 -
578 - errCh := make(chan error, 2) // we only report the 1st error
579 - go func() {
580 - defer func() {
581 - if err := pipeW.Close(); err != nil {
582 - errCh <- fmt.Errorf("stream flush failed: %s", err)
583 - }
584 - close(errCh)
585 - }()
586 -
587 - if err := gocar.WriteCar(
588 - req.Context,
589 - mdag.NewSession(
590 - req.Context,
591 - api.Dag(),
592 - ),
593 - []cid.Cid{c},
594 - pipeW,
595 - ); err != nil {
596 - errCh <- err
597 - }
598 - }()
599 -
600 - if err := res.Emit(pipeR); err != nil {
601 - pipeR.Close() // ignore the error if any
602 - return err
603 - }
604 -
605 - err = <-errCh
606 -
607 - // minimal user friendliness
608 - if err != nil &&
609 - err == ipld.ErrNotFound {
610 - explicitOffline, _ := req.Options["offline"].(bool)
611 - if explicitOffline {
612 - err = fmt.Errorf("%s (currently offline, perhaps retry without the offline flag)", err)
613 - } else {
614 - node, envErr := cmdenv.GetNode(env)
615 - if envErr == nil && !node.IsOnline {
616 - err = fmt.Errorf("%s (currently offline, perhaps retry after attaching to the network)", err)
617 - }
618 - }
619 - }
620 -
621 - return err
622 - },
244 + Run: dagExport,
245 PostRun: cmds.PostRunMap{
624 - cmds.CLI: func(res cmds.Response, re cmds.ResponseEmitter) error {
625 -
626 - var showProgress bool
627 - val, specified := res.Request().Options[progressOptionName]
628 - if !specified {
629 - // default based on TTY availability
630 - errStat, _ := os.Stderr.Stat()
631 - if 0 != (errStat.Mode() & os.ModeCharDevice) {
632 - showProgress = true
633 - }
634 - } else if val.(bool) {
635 - showProgress = true
636 - }
637 -
638 - // simple passthrough, no progress
639 - if !showProgress {
640 - return cmds.Copy(re, res)
641 - }
642 -
643 - bar := pb.New64(0).SetUnits(pb.U_BYTES)
644 - bar.Output = os.Stderr
645 - bar.ShowSpeed = true
646 - bar.ShowElapsedTime = true
647 - bar.RefreshRate = 500 * time.Millisecond
648 - bar.Start()
649 -
650 - var processedOneResponse bool
651 - for {
652 - v, err := res.Next()
653 - if err == io.EOF {
654 -
655 - // We only write the final bar update on success
656 - // On error it looks too weird
657 - bar.Finish()
658 -
659 - return re.Close()
660 - } else if err != nil {
661 - return re.CloseWithError(err)
662 - } else if processedOneResponse {
663 - return re.CloseWithError(errors.New("unexpected multipart response during emit, please file a bugreport"))
664 - }
665 -
666 - r, ok := v.(io.Reader)
667 - if !ok {
668 - // some sort of encoded response, this should not be happening
669 - return errors.New("unexpected non-stream passed to PostRun: please file a bugreport")
670 - }
671 -
672 - processedOneResponse = true
673 -
674 - if err := re.Emit(bar.NewProxyReader(r)); err != nil {
675 - return err
676 - }
677 - }
678 - },
246 + cmds.CLI: finishCLIExport,
247 },
248 }
249
250 +// DagStat is a dag stat command response
251 type DagStat struct {
252 Size uint64
253 NumBlocks int64
@@ -688,6 +257,7 @@ func (s *DagStat) String() string {
257 return fmt.Sprintf("Size: %d, NumBlocks: %d", s.Size, s.NumBlocks)
258 }
259
260 +// DagStatCmd is a command for getting size information about an ipfs-stored dag
261 var DagStatCmd = &cmds.Command{
262 Helptext: cmds.HelpText{
263 Tagline: "Gets stats for a DAG",
@@ -704,81 +274,10 @@ Note: This command skips duplicate blocks in reporting both size and the number
274 Options: []cmds.Option{
275 cmds.BoolOption(progressOptionName, "p", "Return progressive data while reading through the DAG").WithDefault(true),
276 },
707 - Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
708 - progressive := req.Options[progressOptionName].(bool)
709 -
710 - api, err := cmdenv.GetApi(env, req)
711 - if err != nil {
712 - return err
713 - }
714 -
715 - rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
716 - if err != nil {
717 - return err
718 - }
719 -
720 - if len(rp.Remainder()) > 0 {
721 - return fmt.Errorf("cannot return size for anything other than a DAG with a root CID")
722 - }
723 -
724 - nodeGetter := mdag.NewSession(req.Context, api.Dag())
725 - obj, err := nodeGetter.Get(req.Context, rp.Cid())
726 - if err != nil {
727 - return err
728 - }
729 -
730 - dagstats := &DagStat{}
731 - err = traverse.Traverse(obj, traverse.Options{
732 - DAG: nodeGetter,
733 - Order: traverse.DFSPre,
734 - Func: func(current traverse.State) error {
735 - dagstats.Size += uint64(len(current.Node.RawData()))
736 - dagstats.NumBlocks++
737 -
738 - if progressive {
739 - if err := res.Emit(dagstats); err != nil {
740 - return err
741 - }
742 - }
743 - return nil
744 - },
745 - ErrFunc: nil,
746 - SkipDuplicates: true,
747 - })
748 - if err != nil {
749 - return fmt.Errorf("error traversing DAG: %w", err)
750 - }
751 -
752 - if !progressive {
753 - if err := res.Emit(dagstats); err != nil {
754 - return err
755 - }
756 - }
757 -
758 - return nil
759 - },
277 + Run: dagStat,
278 Type: DagStat{},
279 PostRun: cmds.PostRunMap{
762 - cmds.CLI: func(res cmds.Response, re cmds.ResponseEmitter) error {
763 - var dagStats *DagStat
764 - for {
765 - v, err := res.Next()
766 - if err != nil {
767 - if err == io.EOF {
768 - break
769 - }
770 - return err
771 - }
772 -
773 - out, ok := v.(*DagStat)
774 - if !ok {
775 - return e.TypeErr(out, v)
776 - }
777 - dagStats = out
778 - fmt.Fprintf(os.Stderr, "%v\r", out)
779 - }
780 - return re.Emit(dagStats)
781 - },
280 + cmds.CLI: finishCLIStat,
281 },
282 Encoders: cmds.EncoderMap{
283 cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, event *DagStat) error {
core/commands/dag/export.go new
+155
@@ -0,0 +1,155 @@
1 +package dagcmd
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "io"
7 + "os"
8 + "time"
9 +
10 + "github.com/cheggaaa/pb"
11 + cid "github.com/ipfs/go-cid"
12 + "github.com/ipfs/go-ipfs/core/commands/cmdenv"
13 + ipld "github.com/ipfs/go-ipld-format"
14 + mdag "github.com/ipfs/go-merkledag"
15 +
16 + cmds "github.com/ipfs/go-ipfs-cmds"
17 + gocar "github.com/ipld/go-car"
18 +)
19 +
20 +func dagExport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
21 +
22 + c, err := cid.Decode(req.Arguments[0])
23 + if err != nil {
24 + return fmt.Errorf(
25 + "unable to parse root specification (currently only bare CIDs are supported): %s",
26 + err,
27 + )
28 + }
29 +
30 + api, err := cmdenv.GetApi(env, req)
31 + if err != nil {
32 + return err
33 + }
34 +
35 + // Code disabled until descent-issue in go-ipld-prime is fixed
36 + // https://github.com/ribasushi/gip-muddle-up
37 + //
38 + // sb := gipselectorbuilder.NewSelectorSpecBuilder(gipfree.NodeBuilder())
39 + // car := gocar.NewSelectiveCar(
40 + // req.Context,
41 + // <needs to be fixed to take format.NodeGetter as well>,
42 + // []gocar.Dag{gocar.Dag{
43 + // Root: c,
44 + // Selector: sb.ExploreRecursive(
45 + // gipselector.RecursionLimitNone(),
46 + // sb.ExploreAll(sb.ExploreRecursiveEdge()),
47 + // ).Node(),
48 + // }},
49 + // )
50 + // ...
51 + // if err := car.Write(pipeW); err != nil {}
52 +
53 + pipeR, pipeW := io.Pipe()
54 +
55 + errCh := make(chan error, 2) // we only report the 1st error
56 + go func() {
57 + defer func() {
58 + if err := pipeW.Close(); err != nil {
59 + errCh <- fmt.Errorf("stream flush failed: %s", err)
60 + }
61 + close(errCh)
62 + }()
63 +
64 + if err := gocar.WriteCar(
65 + req.Context,
66 + mdag.NewSession(
67 + req.Context,
68 + api.Dag(),
69 + ),
70 + []cid.Cid{c},
71 + pipeW,
72 + ); err != nil {
73 + errCh <- err
74 + }
75 + }()
76 +
77 + if err := res.Emit(pipeR); err != nil {
78 + pipeR.Close() // ignore the error if any
79 + return err
80 + }
81 +
82 + err = <-errCh
83 +
84 + // minimal user friendliness
85 + if err != nil &&
86 + err == ipld.ErrNotFound {
87 + explicitOffline, _ := req.Options["offline"].(bool)
88 + if explicitOffline {
89 + err = fmt.Errorf("%s (currently offline, perhaps retry without the offline flag)", err)
90 + } else {
91 + node, envErr := cmdenv.GetNode(env)
92 + if envErr == nil && !node.IsOnline {
93 + err = fmt.Errorf("%s (currently offline, perhaps retry after attaching to the network)", err)
94 + }
95 + }
96 + }
97 +
98 + return err
99 +}
100 +
101 +func finishCLIExport(res cmds.Response, re cmds.ResponseEmitter) error {
102 +
103 + var showProgress bool
104 + val, specified := res.Request().Options[progressOptionName]
105 + if !specified {
106 + // default based on TTY availability
107 + errStat, _ := os.Stderr.Stat()
108 + if 0 != (errStat.Mode() & os.ModeCharDevice) {
109 + showProgress = true
110 + }
111 + } else if val.(bool) {
112 + showProgress = true
113 + }
114 +
115 + // simple passthrough, no progress
116 + if !showProgress {
117 + return cmds.Copy(re, res)
118 + }
119 +
120 + bar := pb.New64(0).SetUnits(pb.U_BYTES)
121 + bar.Output = os.Stderr
122 + bar.ShowSpeed = true
123 + bar.ShowElapsedTime = true
124 + bar.RefreshRate = 500 * time.Millisecond
125 + bar.Start()
126 +
127 + var processedOneResponse bool
128 + for {
129 + v, err := res.Next()
130 + if err == io.EOF {
131 +
132 + // We only write the final bar update on success
133 + // On error it looks too weird
134 + bar.Finish()
135 +
136 + return re.Close()
137 + } else if err != nil {
138 + return re.CloseWithError(err)
139 + } else if processedOneResponse {
140 + return re.CloseWithError(errors.New("unexpected multipart response during emit, please file a bugreport"))
141 + }
142 +
143 + r, ok := v.(io.Reader)
144 + if !ok {
145 + // some sort of encoded response, this should not be happening
146 + return errors.New("unexpected non-stream passed to PostRun: please file a bugreport")
147 + }
148 +
149 + processedOneResponse = true
150 +
151 + if err := re.Emit(bar.NewProxyReader(r)); err != nil {
152 + return err
153 + }
154 + }
155 +}
core/commands/dag/get.go new
+38
@@ -0,0 +1,38 @@
1 +package dagcmd
2 +
3 +import (
4 + "strings"
5 +
6 + "github.com/ipfs/go-ipfs/core/commands/cmdenv"
7 + "github.com/ipfs/interface-go-ipfs-core/path"
8 +
9 + cmds "github.com/ipfs/go-ipfs-cmds"
10 +)
11 +
12 +func dagGet(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
13 + api, err := cmdenv.GetApi(env, req)
14 + if err != nil {
15 + return err
16 + }
17 +
18 + rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
19 + if err != nil {
20 + return err
21 + }
22 +
23 + obj, err := api.Dag().Get(req.Context, rp.Cid())
24 + if err != nil {
25 + return err
26 + }
27 +
28 + var out interface{} = obj
29 + if len(rp.Remainder()) > 0 {
30 + rem := strings.Split(rp.Remainder(), "/")
31 + final, _, err := obj.Resolve(rem)
32 + if err != nil {
33 + return err
34 + }
35 + out = final
36 + }
37 + return cmds.EmitOnce(res, &out)
38 +}
core/commands/dag/import.go new
+201
@@ -0,0 +1,201 @@
1 +package dagcmd
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "io"
7 +
8 + cid "github.com/ipfs/go-cid"
9 + files "github.com/ipfs/go-ipfs-files"
10 + "github.com/ipfs/go-ipfs/core/commands/cmdenv"
11 + ipld "github.com/ipfs/go-ipld-format"
12 + iface "github.com/ipfs/interface-go-ipfs-core"
13 + "github.com/ipfs/interface-go-ipfs-core/options"
14 +
15 + cmds "github.com/ipfs/go-ipfs-cmds"
16 + gocar "github.com/ipld/go-car"
17 +)
18 +
19 +func dagImport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
20 +
21 + node, err := cmdenv.GetNode(env)
22 + if err != nil {
23 + return err
24 + }
25 +
26 + api, err := cmdenv.GetApi(env, req)
27 + if err != nil {
28 + return err
29 + }
30 +
31 + // on import ensure we do not reach out to the network for any reason
32 + // if a pin based on what is imported + what is in the blockstore
33 + // isn't possible: tough luck
34 + api, err = api.WithOptions(options.Api.Offline(true))
35 + if err != nil {
36 + return err
37 + }
38 +
39 + // grab a pinlock ( which doubles as a GC lock ) so that regardless of the
40 + // size of the streamed-in cars nothing will disappear on us before we had
41 + // a chance to roots that may show up at the very end
42 + // This is especially important for use cases like dagger:
43 + // ipfs dag import $( ... | ipfs-dagger --stdout=carfifos )
44 + //
45 + unlocker := node.Blockstore.PinLock()
46 + defer unlocker.Unlock()
47 +
48 + doPinRoots, _ := req.Options[pinRootsOptionName].(bool)
49 +
50 + retCh := make(chan importResult, 1)
51 + go importWorker(req, res, api, retCh)
52 +
53 + done := <-retCh
54 + if done.err != nil {
55 + return done.err
56 + }
57 +
58 + // It is not guaranteed that a root in a header is actually present in the same ( or any )
59 + // .car file. This is the case in version 1, and ideally in further versions too
60 + // Accumulate any root CID seen in a header, and supplement its actual node if/when encountered
61 + // We will attempt a pin *only* at the end in case all car files were well formed
62 + //
63 + // The boolean value indicates whether we have encountered the root within the car file's
64 + roots := done.roots
65 +
66 + // opportunistic pinning: try whatever sticks
67 + if doPinRoots {
68 +
69 + var failedPins int
70 + for c := range roots {
71 +
72 + // We need to re-retrieve a block, convert it to ipld, and feed it
73 + // to the Pinning interface, sigh...
74 + //
75 + // If we didn't have the problem of inability to take multiple pinlocks,
76 + // we could use the api directly like so (though internally it does the same):
77 + //
78 + // // not ideal, but the pinning api takes only paths :(
79 + // rp := path.NewResolvedPath(
80 + // ipfspath.FromCid(c),
81 + // c,
82 + // c,
83 + // "",
84 + // )
85 + //
86 + // if err := api.Pin().Add(req.Context, rp, options.Pin.Recursive(true)); err != nil {
87 +
88 + ret := RootMeta{Cid: c}
89 +
90 + if block, err := node.Blockstore.Get(c); err != nil {
91 + ret.PinErrorMsg = err.Error()
92 + } else if nd, err := ipld.Decode(block); err != nil {
93 + ret.PinErrorMsg = err.Error()
94 + } else if err := node.Pinning.Pin(req.Context, nd, true); err != nil {
95 + ret.PinErrorMsg = err.Error()
96 + } else if err := node.Pinning.Flush(req.Context); err != nil {
97 + ret.PinErrorMsg = err.Error()
98 + }
99 +
100 + if ret.PinErrorMsg != "" {
101 + failedPins++
102 + }
103 +
104 + if err := res.Emit(&CarImportOutput{Root: ret}); err != nil {
105 + return err
106 + }
107 + }
108 +
109 + if failedPins > 0 {
110 + return fmt.Errorf(
111 + "unable to pin all roots: %d out of %d failed",
112 + failedPins,
113 + len(roots),
114 + )
115 + }
116 + }
117 +
118 + return nil
119 +}
120 +
121 +func importWorker(req *cmds.Request, re cmds.ResponseEmitter, api iface.CoreAPI, ret chan importResult) {
122 +
123 + // this is *not* a transaction
124 + // it is simply a way to relieve pressure on the blockstore
125 + // similar to pinner.Pin/pinner.Flush
126 + batch := ipld.NewBatch(req.Context, api.Dag())
127 +
128 + roots := make(map[cid.Cid]struct{})
129 +
130 + it := req.Files.Entries()
131 + for it.Next() {
132 +
133 + file := files.FileFromEntry(it)
134 + if file == nil {
135 + ret <- importResult{err: errors.New("expected a file handle")}
136 + return
137 + }
138 +
139 + // wrap a defer-closer-scope
140 + //
141 + // every single file in it() is already open before we start
142 + // just close here sooner rather than later for neatness
143 + // and to surface potential errors writing on closed fifos
144 + // this won't/can't help with not running out of handles
145 + err := func() error {
146 + defer file.Close()
147 +
148 + car, err := gocar.NewCarReader(file)
149 + if err != nil {
150 + return err
151 + }
152 +
153 + // Be explicit here, until the spec is finished
154 + if car.Header.Version != 1 {
155 + return errors.New("only car files version 1 supported at present")
156 + }
157 +
158 + for _, c := range car.Header.Roots {
159 + roots[c] = struct{}{}
160 + }
161 +
162 + for {
163 + block, err := car.Next()
164 + if err != nil && err != io.EOF {
165 + return err
166 + } else if block == nil {
167 + break
168 + }
169 +
170 + // the double-decode is suboptimal, but we need it for batching
171 + nd, err := ipld.Decode(block)
172 + if err != nil {
173 + return err
174 + }
175 +
176 + if err := batch.Add(req.Context, nd); err != nil {
177 + return err
178 + }
179 + }
180 +
181 + return nil
182 + }()
183 +
184 + if err != nil {
185 + ret <- importResult{err: err}
186 + return
187 + }
188 + }
189 +
190 + if err := it.Err(); err != nil {
191 + ret <- importResult{err: err}
192 + return
193 + }
194 +
195 + if err := batch.Commit(); err != nil {
196 + ret <- importResult{err: err}
197 + return
198 + }
199 +
200 + ret <- importResult{roots: roots}
201 +}
core/commands/dag/put.go new
+80
@@ -0,0 +1,80 @@
1 +package dagcmd
2 +
3 +import (
4 + "fmt"
5 + "math"
6 +
7 + "github.com/ipfs/go-ipfs/core/commands/cmdenv"
8 + "github.com/ipfs/go-ipfs/core/coredag"
9 +
10 + cmds "github.com/ipfs/go-ipfs-cmds"
11 + files "github.com/ipfs/go-ipfs-files"
12 + ipld "github.com/ipfs/go-ipld-format"
13 + mh "github.com/multiformats/go-multihash"
14 +)
15 +
16 +func dagPut(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
17 + api, err := cmdenv.GetApi(env, req)
18 + if err != nil {
19 + return err
20 + }
21 +
22 + ienc, _ := req.Options["input-enc"].(string)
23 + format, _ := req.Options["format"].(string)
24 + hash, _ := req.Options["hash"].(string)
25 + dopin, _ := req.Options["pin"].(bool)
26 +
27 + // mhType tells inputParser which hash should be used. MaxUint64 means 'use
28 + // default hash' (sha256 for cbor, sha1 for git..)
29 + mhType := uint64(math.MaxUint64)
30 +
31 + if hash != "" {
32 + var ok bool
33 + mhType, ok = mh.Names[hash]
34 + if !ok {
35 + return fmt.Errorf("%s in not a valid multihash name", hash)
36 + }
37 + }
38 +
39 + var adder ipld.NodeAdder = api.Dag()
40 + if dopin {
41 + adder = api.Dag().Pinning()
42 + }
43 + b := ipld.NewBatch(req.Context, adder)
44 +
45 + it := req.Files.Entries()
46 + for it.Next() {
47 + file := files.FileFromEntry(it)
48 + if file == nil {
49 + return fmt.Errorf("expected a regular file")
50 + }
51 + nds, err := coredag.ParseInputs(ienc, format, file, mhType, -1)
52 + if err != nil {
53 + return err
54 + }
55 + if len(nds) == 0 {
56 + return fmt.Errorf("no node returned from ParseInputs")
57 + }
58 +
59 + for _, nd := range nds {
60 + err := b.Add(req.Context, nd)
61 + if err != nil {
62 + return err
63 + }
64 + }
65 +
66 + cid := nds[0].Cid()
67 + if err := res.Emit(&OutputObject{Cid: cid}); err != nil {
68 + return err
69 + }
70 + }
71 + if it.Err() != nil {
72 + return it.Err()
73 + }
74 +
75 + if err := b.Commit(); err != nil {
76 + return err
77 + }
78 +
79 + return nil
80 +}
core/commands/dag/resolve.go new
+25
@@ -0,0 +1,25 @@
1 +package dagcmd
2 +
3 +import (
4 + "github.com/ipfs/go-ipfs/core/commands/cmdenv"
5 + "github.com/ipfs/interface-go-ipfs-core/path"
6 +
7 + cmds "github.com/ipfs/go-ipfs-cmds"
8 +)
9 +
10 +func dagResolve(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
11 + api, err := cmdenv.GetApi(env, req)
12 + if err != nil {
13 + return err
14 + }
15 +
16 + rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
17 + if err != nil {
18 + return err
19 + }
20 +
21 + return cmds.EmitOnce(res, &ResolveOutput{
22 + Cid: rp.Cid(),
23 + RemPath: rp.Remainder(),
24 + })
25 +}
core/commands/dag/stat.go new
+90
@@ -0,0 +1,90 @@
1 +package dagcmd
2 +
3 +import (
4 + "fmt"
5 + "io"
6 + "os"
7 +
8 + "github.com/ipfs/go-ipfs/core/commands/cmdenv"
9 + "github.com/ipfs/go-ipfs/core/commands/e"
10 + "github.com/ipfs/go-merkledag/traverse"
11 + "github.com/ipfs/interface-go-ipfs-core/path"
12 +
13 + cmds "github.com/ipfs/go-ipfs-cmds"
14 + mdag "github.com/ipfs/go-merkledag"
15 +)
16 +
17 +func dagStat(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
18 + progressive := req.Options[progressOptionName].(bool)
19 +
20 + api, err := cmdenv.GetApi(env, req)
21 + if err != nil {
22 + return err
23 + }
24 +
25 + rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
26 + if err != nil {
27 + return err
28 + }
29 +
30 + if len(rp.Remainder()) > 0 {
31 + return fmt.Errorf("cannot return size for anything other than a DAG with a root CID")
32 + }
33 +
34 + nodeGetter := mdag.NewSession(req.Context, api.Dag())
35 + obj, err := nodeGetter.Get(req.Context, rp.Cid())
36 + if err != nil {
37 + return err
38 + }
39 +
40 + dagstats := &DagStat{}
41 + err = traverse.Traverse(obj, traverse.Options{
42 + DAG: nodeGetter,
43 + Order: traverse.DFSPre,
44 + Func: func(current traverse.State) error {
45 + dagstats.Size += uint64(len(current.Node.RawData()))
46 + dagstats.NumBlocks++
47 +
48 + if progressive {
49 + if err := res.Emit(dagstats); err != nil {
50 + return err
51 + }
52 + }
53 + return nil
54 + },
55 + ErrFunc: nil,
56 + SkipDuplicates: true,
57 + })
58 + if err != nil {
59 + return fmt.Errorf("error traversing DAG: %w", err)
60 + }
61 +
62 + if !progressive {
63 + if err := res.Emit(dagstats); err != nil {
64 + return err
65 + }
66 + }
67 +
68 + return nil
69 +}
70 +
71 +func finishCLIStat(res cmds.Response, re cmds.ResponseEmitter) error {
72 + var dagStats *DagStat
73 + for {
74 + v, err := res.Next()
75 + if err != nil {
76 + if err == io.EOF {
77 + break
78 + }
79 + return err
80 + }
81 +
82 + out, ok := v.(*DagStat)
83 + if !ok {
84 + return e.TypeErr(out, v)
85 + }
86 + dagStats = out
87 + fmt.Fprintf(os.Stderr, "%v\r", out)
88 + }
89 + return re.Emit(dagStats)
90 +}