implement publisher for ipns to wait until moments of rapid churn die down

fuse/ipns/ipns_unix.go

@@ -76,6 +76,9 @@ func CreateRoot(n *core.IpfsNode, keys []ci.PrivKey, ipfsroot string) (*Root, er
 		nd := new(Node)
 		nd.Ipfs = n
 		nd.key = k
+		nd.repub = NewRepublisher(nd, time.Millisecond*10)
+
+		go nd.repub.Run()
 
 		pointsTo, err := n.Namesys.Resolve(name)
 		if err != nil {
@@ -185,6 +188,8 @@ func (r *Root) ReadDir(intr fs.Intr) ([]fuse.Dirent, fuse.Error) {
 type Node struct {
 	nsRoot *Node
 
+	repub *Republisher
+
 	// Name really only for logging purposes
 	name string
 
@@ -316,14 +321,18 @@ func (n *Node) Flush(req *fuse.FlushRequest, intr fs.Intr) fuse.Error {
 			return fuse.ENODATA
 		}
 
-		err = n.updateTree()
-		if err != nil {
-			log.Error("updateTree failed: %s", err)
-			return fuse.ENODATA
+		n.sendPublishSignal()
 	}
+	return nil
+}
 
+func (n *Node) sendPublishSignal() {
+	root := n.nsRoot
+	if root == nil {
+		root = n
 	}
-	return nil
+
+	root.repub.Publish <- struct{}{}
 }
 
 func (n *Node) updateTree() error {
@@ -387,7 +396,7 @@ func (n *Node) Mkdir(req *fuse.MkdirRequest, intr fs.Intr) (fs.Node, fuse.Error)
 	}
 
 	n.changed = true
-	n.updateTree()
+	n.sendPublishSignal()
 
 	return child, nil
 }
@@ -425,6 +434,8 @@ func (n *Node) Remove(req *fuse.RemoveRequest, intr fs.Intr) fuse.Error {
 		log.Error("Remove: No such file.")
 		return fuse.ENOENT
 	}
+	n.changed = true
+	n.sendPublishSignal()
 	return nil
 }
 
@@ -471,7 +482,7 @@ func Mount(ipfs *core.IpfsNode, fpath string, ipfspath string) error {
 			if err == nil {
 				return
 			}
-			time.Sleep(time.Millisecond * 10)
+			time.Sleep(time.Millisecond * 100)
 		}
 		ipfs.Network.Close()
 	}()
@@ -563,12 +574,13 @@ func NewRepublisher(n *Node, tout time.Duration) *Republisher {
 }
 
 func (np *Republisher) Run() {
-	for _ := range np.Publish {
+	for _ = range np.Publish {
 		timer := time.After(np.Timeout)
 		for {
 			select {
 			case <-timer:
 				//Do the publish!
+				log.Info("Publishing Changes!")
 				err := np.node.updateTree()
 				if err != nil {
 					log.Critical("updateTree error: %s", err)

routing/dht/dht.go

@@ -177,8 +177,7 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p *peer.Peer, pmes *Message
 	start := time.Now()
 
 	// Print out diagnostic
-	log.Debug("[peer: %s] Sent message type: '%s' [to = %s]\n",
-		dht.self.ID.Pretty(),
+	log.Debug("Sent message type: '%s' [to = %s]",
 		Message_MessageType_name[int32(pmes.GetType())], p.ID.Pretty())
 
 	rmes, err := dht.sender.SendRequest(ctx, mes)
@@ -281,7 +280,7 @@ func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p *peer.Peer,
 	}
 
 	if len(peers) > 0 {
-		u.DOut("getValueOrPeers: peers")
+		log.Debug("getValueOrPeers: peers")
 		return nil, peers, nil
 	}
 
@@ -400,7 +399,7 @@ func (dht *IpfsDHT) addProviders(key u.Key, peers []*Message_Peer) []*peer.Peer
 	for _, prov := range peers {
 		p, err := dht.peerFromInfo(prov)
 		if err != nil {
-			u.PErr("error getting peer from info: %v\n", err)
+			log.Error("error getting peer from info: %v", err)
 			continue
 		}
 

routing/dht/routing.go

@@ -18,7 +18,7 @@ import (
 // PutValue adds value corresponding to given Key.
 // This is the top level "Store" operation of the DHT
 func (dht *IpfsDHT) PutValue(ctx context.Context, key u.Key, value []byte) error {
-	log.Debug("PutValue %s %v", key.Pretty(), value)
+	log.Debug("PutValue %s", key.Pretty())
 	err := dht.putLocal(key, value)
 	if err != nil {
 		return err