flush pinning improvements

License: MIT
Signed-off-by: Jeromy <jeromyj@gmail.com>

core/core.go

@@ -483,8 +483,11 @@ func (n *IpfsNode) loadFilesRoot() error {
 			return err
 		}
 
-		err = n.Pinning.Pin(n.Context(), nnd, true)
+		if err := n.Pinning.Pin(n.Context(), nnd, true); err != nil {
-		if err != nil {
+			return err
+		}
+
+		if err := n.Pinning.Flush(); err != nil {
 			return err
 		}
 

mfs/ops.go

@@ -197,6 +197,9 @@ func DirLookup(d *Directory, pth string) (FSNode, error) {
 
 func FlushPath(r *Root, pth string) error {
 	parts := path.SplitList(strings.Trim(pth, "/"))
+	if len(parts) == 1 && parts[0] == "" {
+		parts = nil
+	}
 
 	d, ok := r.GetValue().(*Directory)
 	if !ok {
@@ -214,12 +217,24 @@ func FlushPath(r *Root, pth string) error {
 	}
 
 	r.repub.Update(k)
+	r.repub.WaitPub()
+
 	return nil
 }
 
 func flushPathRec(d *Directory, parts []string) (*dag.Node, error) {
 	if len(parts) == 0 {
-		return d.GetNode()
+		nd, err := d.GetNode()
+		if err != nil {
+			return nil, err
+		}
+
+		_, err = d.dserv.Add(nd)
+		if err != nil {
+			return nil, err
+		}
+
+		return nd, nil
 	}
 
 	d.Lock()
@@ -243,6 +258,11 @@ func flushPathRec(d *Directory, parts []string) (*dag.Node, error) {
 			return nil, err
 		}
 
+		_, err = d.dserv.Add(newnode)
+		if err != nil {
+			return nil, err
+		}
+
 		d.node = newnode
 		return newnode, nil
 	case *File:

mfs/system.go

@@ -165,7 +165,7 @@ type Republisher struct {
 	TimeoutShort time.Duration
 	Publish      chan struct{}
 	pubfunc      PubFunc
-	pubnowch     chan struct{}
+	pubnowch     chan chan struct{}
 
 	ctx    context.Context
 	cancel func()
@@ -190,7 +190,7 @@ func NewRepublisher(ctx context.Context, pf PubFunc, tshort, tlong time.Duration
 		TimeoutLong:  tlong,
 		Publish:      make(chan struct{}, 1),
 		pubfunc:      pf,
-		pubnowch:     make(chan struct{}),
+		pubnowch:     make(chan chan struct{}),
 		ctx:          ctx,
 		cancel:       cancel,
 	}
@@ -204,11 +204,17 @@ func (p *Republisher) setVal(k key.Key) {
 
 func (p *Republisher) pubNow() {
 	select {
-	case p.pubnowch <- struct{}{}:
+	case p.pubnowch <- nil:
 	default:
 	}
 }
 
+func (p *Republisher) WaitPub() {
+	wait := make(chan struct{})
+	p.pubnowch <- wait
+	<-wait
+}
+
 func (p *Republisher) Close() error {
 	err := p.publish(p.ctx)
 	p.cancel()
@@ -235,6 +241,8 @@ func (np *Republisher) Run() {
 			longer := time.After(np.TimeoutLong)
 
 		wait:
+			var pubnowresp chan struct{}
+
 			select {
 			case <-np.ctx.Done():
 				return
@@ -243,10 +251,13 @@ func (np *Republisher) Run() {
 				goto wait
 			case <-quick:
 			case <-longer:
-			case <-np.pubnowch:
+			case pubnowresp = <-np.pubnowch:
 			}
 
 			err := np.publish(np.ctx)
+			if pubnowresp != nil {
+				pubnowresp <- struct{}{}
+			}
 			if err != nil {
 				log.Error("republishRoot error: %s", err)
 			}

test/sharness/t0250-files-api.sh

@@ -336,9 +336,13 @@ test_files_api() {
 		test_cmp root_hash_exp root_hash
 	'
 
-	test_expect_success "root hash is pinned" '
+	test_expect_success "flush root succeeds" '
-		ipfs pin ls
+		ipfs files flush /
-		return 1
+	'
+
+	test_expect_success "root hash is pinned after flush" '
+		ipfs pin ls > pins &&
+			grep $EXP_ROOT_HASH pins || (cat pins && exit 1)
 	'
 
 	# test mv