feat(ocis): use only one channel for backup command

Signed-off-by: jkoberg <jkoberg@owncloud.com>

.drone.star

@@ -1989,7 +1989,6 @@ def ocisServer(storage, accounts_hash_difficulty = 4, volumes = [], depends_on =
             "commands": [
                 "%s init --insecure true" % ocis_bin,
                 "cat $OCIS_CONFIG_DIR/ocis.yaml",
-                "%s backup consistency -p /var/lib/ocis/storage/users" % ocis_bin,
             ] + (wrapper_commands),
             "volumes": volumes,
             "depends_on": depends_on,

ocis/pkg/backup/backup.go

@@ -69,14 +69,53 @@ func CheckProviderConsistency(storagepath string, lbs ListBlobstore) error {
 	}
 
 	c := NewConsistency()
-	c.GatherData(p.Nodes, p.Links, p.Blobs, p.Quit)
+	c.GatherData(p.Events)
 
 	return c.PrintResults(storagepath)
 }
 
 // GatherData gathers and evaluates data produced by the DataProvider
-func (c *Consistency) GatherData(nodes <-chan NodeData, links <-chan LinkData, blobs <-chan BlobData, quit <-chan struct{}) {
-	c.gatherData(nodes, links, blobs, quit)
+func (c *Consistency) GatherData(events <-chan interface{}) {
+	for ev := range events {
+		switch d := ev.(type) {
+		case NodeData:
+			// does it have inconsistencies?
+			if len(d.Inconsistencies) != 0 {
+				c.Nodes[d.NodePath] = append(c.Nodes[d.NodePath], d.Inconsistencies...)
+			}
+			// is it linked?
+			if _, ok := c.LinkedNodes[d.NodePath]; ok {
+				deleteInconsistency(c.LinkedNodes, d.NodePath)
+			} else if d.RequiresSymlink && c.Nodes[d.NodePath] == nil {
+				c.Nodes[d.NodePath] = []Inconsistency{}
+			}
+			// does it have a blob?
+			if d.BlobPath != "" {
+				if _, ok := c.Blobs[d.BlobPath]; ok {
+					deleteInconsistency(c.Blobs, d.BlobPath)
+				} else {
+					c.BlobReferences[d.BlobPath] = []Inconsistency{}
+					c.blobToNode[d.BlobPath] = d.NodePath
+				}
+			}
+		case LinkData:
+			// does it have a node?
+			if _, ok := c.Nodes[d.NodePath]; ok {
+				deleteInconsistency(c.Nodes, d.NodePath)
+			} else {
+				c.LinkedNodes[d.NodePath] = []Inconsistency{}
+				c.nodeToLink[d.NodePath] = d.LinkPath
+			}
+		case BlobData:
+			// does it have a reference?
+			if _, ok := c.BlobReferences[d.BlobPath]; ok {
+				deleteInconsistency(c.BlobReferences, d.BlobPath)
+			} else {
+				c.Blobs[d.BlobPath] = []Inconsistency{}
+			}
+
+		}
+	}
 
 	for n := range c.Nodes {
 		if len(c.Nodes[n]) == 0 {
@@ -94,52 +133,6 @@ func (c *Consistency) GatherData(nodes <-chan NodeData, links <-chan LinkData, b
 	}
 }
 
-func (c *Consistency) gatherData(nodes <-chan NodeData, links <-chan LinkData, blobs <-chan BlobData, quit <-chan struct{}) {
-	for {
-		select {
-		case n := <-nodes:
-			// does it have inconsistencies?
-			if len(n.Inconsistencies) != 0 {
-				c.Nodes[n.NodePath] = append(c.Nodes[n.NodePath], n.Inconsistencies...)
-			}
-			// is it linked?
-			if _, ok := c.LinkedNodes[n.NodePath]; ok {
-				deleteInconsistency(c.LinkedNodes, n.NodePath)
-			} else if n.RequiresSymlink {
-				c.Nodes[n.NodePath] = c.Nodes[n.NodePath]
-			}
-			// does it have a blob?
-			if n.BlobPath != "" {
-				if _, ok := c.Blobs[n.BlobPath]; ok {
-					deleteInconsistency(c.Blobs, n.BlobPath)
-				} else {
-					c.BlobReferences[n.BlobPath] = []Inconsistency{}
-					c.blobToNode[n.BlobPath] = n.NodePath
-				}
-			}
-		case l := <-links:
-			// does it have a node?
-			if _, ok := c.Nodes[l.NodePath]; ok {
-				deleteInconsistency(c.Nodes, l.NodePath)
-			} else {
-				c.LinkedNodes[l.NodePath] = []Inconsistency{}
-				c.nodeToLink[l.NodePath] = l.LinkPath
-			}
-		case b := <-blobs:
-			// does it have a reference?
-			if _, ok := c.BlobReferences[b.BlobPath]; ok {
-				deleteInconsistency(c.BlobReferences, b.BlobPath)
-			} else {
-				c.Blobs[b.BlobPath] = []Inconsistency{}
-			}
-		case <-quit:
-			return
-
-		}
-	}
-
-}
-
 // PrintResults prints the results of the evaluation
 func (c *Consistency) PrintResults(discpath string) error {
 	if len(c.Nodes) != 0 {

ocis/pkg/backup/backup_test.go

@@ -90,31 +90,24 @@ func TestGatherData(t *testing.T) {
 	}
 
 	for _, tc := range testcases {
-		nodes := make(chan backup.NodeData)
-		links := make(chan backup.LinkData)
-		blobs := make(chan backup.BlobData)
-		quit := make(chan struct{})
+		events := make(chan interface{})
 
 		go func() {
 			for _, ev := range tc.Events {
 				switch e := ev.(type) {
 				case backup.NodeData:
-					nodes <- e
+					events <- e
 				case backup.LinkData:
-					links <- e
+					events <- e
 				case backup.BlobData:
-					blobs <- e
+					events <- e
 				}
 			}
-			quit <- struct{}{}
-			close(nodes)
-			close(links)
-			close(blobs)
-			close(quit)
+			close(events)
 		}()
 
 		c := backup.NewConsistency()
-		c.GatherData(nodes, links, blobs, quit)
+		c.GatherData(events)
 
 		require.Equal(t, tc.Expected.Nodes, c.Nodes)
 		require.Equal(t, tc.Expected.LinkedNodes, c.LinkedNodes)

ocis/pkg/backup/provider.go

@@ -21,10 +21,7 @@ type ListBlobstore interface {
 
 // DataProvider provides data for the consistency check
 type DataProvider struct {
-	Nodes chan NodeData
-	Links chan LinkData
-	Blobs chan BlobData
-	Quit  chan struct{}
+	Events chan interface{}
 
 	fsys     fs.FS
 	discpath string
@@ -53,10 +50,7 @@ type BlobData struct {
 // NewProvider creates a new DataProvider object
 func NewProvider(fsys fs.FS, discpath string, lbs ListBlobstore) *DataProvider {
 	return &DataProvider{
-		Nodes: make(chan NodeData),
-		Links: make(chan LinkData),
-		Blobs: make(chan BlobData),
-		Quit:  make(chan struct{}),
+		Events: make(chan interface{}),
 
 		fsys:     fsys,
 		discpath: discpath,
@@ -82,47 +76,7 @@ func (dp *DataProvider) ProduceData() error {
 	wg.Add(1)
 	go func() {
 		for _, d := range dirs {
-			entries, err := fs.ReadDir(dp.fsys, d)
-			if err != nil {
-				fmt.Println("error reading dir", err)
-				continue
-			}
-
-			if len(entries) == 0 {
-				fmt.Println("empty dir", filepath.Join(dp.discpath, d))
-				continue
-			}
-
-			for _, e := range entries {
-				switch {
-				case e.IsDir():
-					ls, err := fs.ReadDir(dp.fsys, filepath.Join(d, e.Name()))
-					if err != nil {
-						fmt.Println("error reading dir", err)
-						continue
-					}
-					for _, l := range ls {
-						linkpath := filepath.Join(dp.discpath, d, e.Name(), l.Name())
-
-						r, _ := os.Readlink(linkpath)
-						nodePath := filepath.Join(dp.discpath, d, e.Name(), r)
-						dp.Links <- LinkData{LinkPath: linkpath, NodePath: nodePath}
-					}
-					fallthrough
-				case filepath.Ext(e.Name()) == "" || _versionRegex.MatchString(e.Name()) || _trashRegex.MatchString(e.Name()):
-					np := filepath.Join(dp.discpath, d, e.Name())
-					var inc []Inconsistency
-					if !dp.filesExist(filepath.Join(d, e.Name())) {
-						inc = append(inc, InconsistencyFilesMissing)
-					}
-					bp, i := dp.getBlobPath(filepath.Join(d, e.Name()))
-					if i != "" {
-						inc = append(inc, i)
-					}
-
-					dp.Nodes <- NodeData{NodePath: np, BlobPath: bp, RequiresSymlink: requiresSymlink(np), Inconsistencies: inc}
-				}
-			}
+			dp.evaluateNodeDir(d)
 		}
 		wg.Done()
 	}()
@@ -130,16 +84,7 @@ func (dp *DataProvider) ProduceData() error {
 	// crawl trash
 	wg.Add(1)
 	go func() {
-		linkpaths, err := fs.Glob(dp.fsys, "spaces/*/*/trash/*/*/*/*/*")
-		if err != nil {
-			fmt.Println("error reading trash", err)
-		}
-		for _, l := range linkpaths {
-			linkpath := filepath.Join(dp.discpath, l)
-			r, _ := os.Readlink(linkpath)
-			p := filepath.Join(dp.discpath, l, "..", r)
-			dp.Links <- LinkData{LinkPath: linkpath, NodePath: p}
-		}
+		dp.evaluateTrashDir()
 		wg.Done()
 	}()
 
@@ -152,7 +97,7 @@ func (dp *DataProvider) ProduceData() error {
 		}
 
 		for _, bn := range bs {
-			dp.Blobs <- BlobData{BlobPath: dp.lbs.Path(bn)}
+			dp.Events <- BlobData{BlobPath: dp.lbs.Path(bn)}
 		}
 		wg.Done()
 	}()
@@ -187,6 +132,65 @@ func (dp *DataProvider) getBlobPath(path string) (string, Inconsistency) {
 	return "", ""
 }
 
+func (dp *DataProvider) evaluateNodeDir(d string) {
+	// d is something like spaces/a8/e5d981-41e4-4468-b532-258d5fb457d3/nodes/2d/08/8d/24
+	// we could have multiple nodes under this, but we are only interested in one file per node - the one with "" extension
+	entries, err := fs.ReadDir(dp.fsys, d)
+	if err != nil {
+		fmt.Println("error reading dir", err)
+		return
+	}
+
+	if len(entries) == 0 {
+		fmt.Println("empty dir", filepath.Join(dp.discpath, d))
+		return
+	}
+
+	for _, e := range entries {
+		switch {
+		case e.IsDir():
+			ls, err := fs.ReadDir(dp.fsys, filepath.Join(d, e.Name()))
+			if err != nil {
+				fmt.Println("error reading dir", err)
+				continue
+			}
+			for _, l := range ls {
+				linkpath := filepath.Join(dp.discpath, d, e.Name(), l.Name())
+
+				r, _ := os.Readlink(linkpath)
+				nodePath := filepath.Join(dp.discpath, d, e.Name(), r)
+				dp.Events <- LinkData{LinkPath: linkpath, NodePath: nodePath}
+			}
+			fallthrough
+		case filepath.Ext(e.Name()) == "" || _versionRegex.MatchString(e.Name()) || _trashRegex.MatchString(e.Name()):
+			np := filepath.Join(dp.discpath, d, e.Name())
+			var inc []Inconsistency
+			if !dp.filesExist(filepath.Join(d, e.Name())) {
+				inc = append(inc, InconsistencyFilesMissing)
+			}
+			bp, i := dp.getBlobPath(filepath.Join(d, e.Name()))
+			if i != "" {
+				inc = append(inc, i)
+			}
+
+			dp.Events <- NodeData{NodePath: np, BlobPath: bp, RequiresSymlink: requiresSymlink(np), Inconsistencies: inc}
+		}
+	}
+}
+
+func (dp *DataProvider) evaluateTrashDir() {
+	linkpaths, err := fs.Glob(dp.fsys, "spaces/*/*/trash/*/*/*/*/*")
+	if err != nil {
+		fmt.Println("error reading trash", err)
+	}
+	for _, l := range linkpaths {
+		linkpath := filepath.Join(dp.discpath, l)
+		r, _ := os.Readlink(linkpath)
+		p := filepath.Join(dp.discpath, l, "..", r)
+		dp.Events <- LinkData{LinkPath: linkpath, NodePath: p}
+	}
+}
+
 func (dp *DataProvider) filesExist(path string) bool {
 	check := func(p string) bool {
 		_, err := fs.Stat(dp.fsys, p)
@@ -196,11 +200,7 @@ func (dp *DataProvider) filesExist(path string) bool {
 }
 
 func (dp *DataProvider) quit() {
-	dp.Quit <- struct{}{}
-	close(dp.Nodes)
-	close(dp.Links)
-	close(dp.Blobs)
-	close(dp.Quit)
+	close(dp.Events)
 }
 
 func requiresSymlink(path string) bool {