pinning: fix pin listings
Jorropo committed
Jun 2, 2023 at 18:16 UTC
a2c66abc52864bcd7bbb099f5959ec66557c73cd
2 files changed
+31
-40
core/commands/pin/pin.go
+13
-10
@@ -342,14 +342,17 @@ Example:
342
}
343
344
// For backward compatibility, we accumulate the pins in the same output type as before.
345
- emit := res.Emit
345
+ var emit func(PinLsOutputWrapper) error
346
lgcList := map[string]PinLsType{}
347
if !stream {
348
- emit = func(v interface{}) error {
349
- obj := v.(*PinLsOutputWrapper)
350
- lgcList[obj.PinLsObject.Cid] = PinLsType{Type: obj.PinLsObject.Type}
348
+ emit = func(v PinLsOutputWrapper) error {
349
+ lgcList[v.PinLsObject.Cid] = PinLsType{Type: v.PinLsObject.Type}
350
return nil
351
}
352
+ } else {
353
+ emit = func(v PinLsOutputWrapper) error {
354
+ return res.Emit(v)
355
+ }
356
}
357
358
if len(req.Arguments) > 0 {
@@ -371,7 +374,7 @@ Example:
374
},
375
Type: &PinLsOutputWrapper{},
376
Encoders: cmds.EncoderMap{
374
- cmds.JSON: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *PinLsOutputWrapper) error {
377
+ cmds.JSON: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out PinLsOutputWrapper) error {
378
stream, _ := req.Options[pinStreamOptionName].(bool)
379
380
enc := json.NewEncoder(w)
@@ -382,7 +385,7 @@ Example:
385
386
return enc.Encode(out.PinLsList)
387
}),
385
- cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *PinLsOutputWrapper) error {
388
+ cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out PinLsOutputWrapper) error {
389
quiet, _ := req.Options[pinQuietOptionName].(bool)
390
stream, _ := req.Options[pinStreamOptionName].(bool)
391
@@ -432,7 +435,7 @@ type PinLsObject struct {
435
Type string `json:",omitempty"`
436
}
437
435
-func pinLsKeys(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit func(value interface{}) error) error {
438
+func pinLsKeys(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit func(value PinLsOutputWrapper) error) error {
439
enc, err := cmdenv.GetCidEncoder(req)
440
if err != nil {
441
return err
@@ -470,7 +473,7 @@ func pinLsKeys(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit fu
473
pinType = "indirect through " + pinType
474
}
475
473
- err = emit(&PinLsOutputWrapper{
476
+ err = emit(PinLsOutputWrapper{
477
PinLsObject: PinLsObject{
478
Type: pinType,
479
Cid: enc.Encode(rp.Cid()),
@@ -484,7 +487,7 @@ func pinLsKeys(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit fu
487
return nil
488
}
489
487
-func pinLsAll(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit func(value interface{}) error) error {
490
+func pinLsAll(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit func(value PinLsOutputWrapper) error) error {
491
enc, err := cmdenv.GetCidEncoder(req)
492
if err != nil {
493
return err
@@ -511,7 +514,7 @@ func pinLsAll(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit fun
514
if err := p.Err(); err != nil {
515
return err
516
}
514
- err = emit(&PinLsOutputWrapper{
517
+ err = emit(PinLsOutputWrapper{
518
PinLsObject: PinLsObject{
519
Type: p.Type(),
520
Cid: enc.Encode(p.Path().Cid()),
core/coreapi/pin.go
+18
-30
@@ -305,6 +305,7 @@ func (api *PinAPI) pinLsAll(ctx context.Context, typeStr string) <-chan coreifac
305
out <- &pinInfo{err: err}
306
return
307
}
308
+ rkeys = append(rkeys, streamedCid.C)
309
}
310
}
311
if typeStr == "direct" || typeStr == "all" {
@@ -319,27 +320,6 @@ func (api *PinAPI) pinLsAll(ctx context.Context, typeStr string) <-chan coreifac
320
}
321
}
322
}
322
- if typeStr == "all" {
323
- walkingSet := cid.NewSet()
324
- for _, k := range rkeys {
325
- err = merkledag.Walk(
326
- ctx, merkledag.GetLinksWithDAG(api.dag), k,
327
- walkingSet.Visit,
328
- merkledag.SkipRoot(), merkledag.Concurrent(),
329
- )
330
- if err != nil {
331
- out <- &pinInfo{err: err}
332
- return
333
- }
334
- }
335
- err = walkingSet.ForEach(func(c cid.Cid) error {
336
- return AddToResultKeys(c, "indirect")
337
- })
338
- if err != nil {
339
- out <- &pinInfo{err: err}
340
- return
341
- }
342
- }
323
if typeStr == "indirect" {
324
// We need to first visit the direct pins that have priority
325
// without emitting them
@@ -358,13 +338,28 @@ func (api *PinAPI) pinLsAll(ctx context.Context, typeStr string) <-chan coreifac
338
return
339
}
340
emittedSet.Add(streamedCid.C)
341
+ rkeys = append(rkeys, streamedCid.C)
342
}
362
-
343
+ }
344
+ if typeStr == "indirect" || typeStr == "all" {
345
walkingSet := cid.NewSet()
346
for _, k := range rkeys {
347
err = merkledag.Walk(
348
ctx, merkledag.GetLinksWithDAG(api.dag), k,
367
- walkingSet.Visit,
349
+ func(c cid.Cid) bool {
350
+ if !walkingSet.Visit(c) {
351
+ return false
352
+ }
353
+ if emittedSet.Has(c) {
354
+ return true // skipped
355
+ }
356
+ err := AddToResultKeys(c, "indirect")
357
+ if err != nil {
358
+ out <- &pinInfo{err: err}
359
+ return false
360
+ }
361
+ return true
362
+ },
363
merkledag.SkipRoot(), merkledag.Concurrent(),
364
)
365
if err != nil {
@@ -372,13 +367,6 @@ func (api *PinAPI) pinLsAll(ctx context.Context, typeStr string) <-chan coreifac
367
return
368
}
369
}
375
- err = emittedSet.ForEach(func(c cid.Cid) error {
376
- return AddToResultKeys(c, "indirect")
377
- })
378
- if err != nil {
379
- out <- &pinInfo{err: err}
380
- return
381
- }
370
}
371
}()
372