Merge pull request #11704 from kobergj/FixStaleNodesCommand

[full-ci][OCISDEV-355] Fix delete stale nodes command

Author: kobergj

changelog/unreleased/bump-reva.md

@@ -4,3 +4,4 @@ bumped reva version
 
 https://github.com/owncloud/ocis/pull/11669
 https://github.com/owncloud/ocis/pull/11693
+https://github.com/owncloud/ocis/pull/11704

changelog/unreleased/fix-stale-nodes-command.md

@@ -0,0 +1,5 @@
+Bugfix: Fix stale nodes command
+
+Fix delete-stale-nodes command to properly remove or roll-back nodes
+
+https://github.com/owncloud/ocis/pull/11704

go.mod

@@ -58,15 +58,15 @@ require (
 	github.com/mna/pigeon v1.3.0
 	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826
 	github.com/nats-io/nats-server/v2 v2.12.0
-	github.com/nats-io/nats.go v1.45.0
+	github.com/nats-io/nats.go v1.46.0
 	github.com/olekukonko/tablewriter v1.1.0
 	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/ginkgo/v2 v2.25.3
 	github.com/onsi/gomega v1.38.2
 	github.com/open-policy-agent/opa v1.6.0
 	github.com/orcaman/concurrent-map v1.0.0
 	github.com/owncloud/libre-graph-api-go v1.0.5-0.20250217093259-fa3804be6c27
-	github.com/owncloud/reva/v2 v2.0.0-20250925134851-02c7681fe2b3
+	github.com/owncloud/reva/v2 v2.0.0-20251001134122-cb98fe521deb
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/xattr v0.4.12
 	github.com/prometheus/client_golang v1.23.2

go.sum

@@ -666,8 +666,8 @@ github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
 github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
 github.com/nats-io/nats-server/v2 v2.12.0 h1:OIwe8jZUqJFrh+hhiyKu8snNib66qsx806OslqJuo74=
 github.com/nats-io/nats-server/v2 v2.12.0/go.mod h1:nr8dhzqkP5E/lDwmn+A2CvQPMd1yDKXQI7iGg3lAvww=
-github.com/nats-io/nats.go v1.45.0 h1:/wGPbnYXDM0pLKFjZTX+2JOw9TQPoIgTFrUaH97giwA=
-github.com/nats-io/nats.go v1.45.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
+github.com/nats-io/nats.go v1.46.0 h1:iUcX+MLT0HHXskGkz+Sg20sXrPtJLsOojMDTDzOHSb8=
+github.com/nats-io/nats.go v1.46.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
 github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
 github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -703,8 +703,8 @@ github.com/orcaman/concurrent-map v1.0.0 h1:I/2A2XPCb4IuQWcQhBhSwGfiuybl/J0ev9HD
 github.com/orcaman/concurrent-map v1.0.0/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
 github.com/owncloud/libre-graph-api-go v1.0.5-0.20250217093259-fa3804be6c27 h1:ID8s5lGBntmrlI6TbDAjTzRyHucn3bVM2wlW+HBplv4=
 github.com/owncloud/libre-graph-api-go v1.0.5-0.20250217093259-fa3804be6c27/go.mod h1:+gT+x62AS9u2Farh9wE2uYmgdvTg0MQgsSI62D+xoRg=
-github.com/owncloud/reva/v2 v2.0.0-20250925134851-02c7681fe2b3 h1:VL/YicK+o8a2u9IbFM5GlH540C04A0L6Ao3pCgctCXw=
-github.com/owncloud/reva/v2 v2.0.0-20250925134851-02c7681fe2b3/go.mod h1:Re3ZSyWo8fgH6hJ3zgLSgZG/jqeo+O8JX3c+AUzKJro=
+github.com/owncloud/reva/v2 v2.0.0-20251001134122-cb98fe521deb h1:wIu3AM0Khs+cw3BJjXnPz7tjJyZ3DbMsX2pwWafvyRI=
+github.com/owncloud/reva/v2 v2.0.0-20251001134122-cb98fe521deb/go.mod h1:JwGg9zW2BmgMdcjKRwzqyoWtSVhFTMlVa4z6UToQgAA=
 github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c h1:rp5dCmg/yLR3mgFuSOe4oEnDDmGLROTvMragMUXpTQw=
 github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c/go.mod h1:X07ZCGwUbLaax7L0S3Tw4hpejzu63ZrrQiUe6W0hcy0=
 github.com/pablodz/inotifywaitgo v0.0.9 h1:njquRbBU7fuwIe5rEvtaniVBjwWzcpdUVptSgzFqZsw=

services/storage-users/pkg/command/uploads.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/fs"
+	"log"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -16,6 +17,7 @@ import (
 	"github.com/urfave/cli/v2"
 
 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/owncloud/ocis/v2/ocis-pkg/config/configlog"
 	"github.com/owncloud/ocis/v2/services/storage-users/pkg/config"
 	"github.com/owncloud/ocis/v2/services/storage-users/pkg/config/parser"
@@ -376,9 +378,18 @@ func DeleteStaleProcessingNodes(cfg *config.Config) *cli.Command {
 				}
 			}
 
+			var stream events.Stream
+			if !dryRun {
+				s, err := event.NewStream(cfg)
+				if err != nil {
+					log.Fatalf("Failed to create event stream: %v", err)
+				}
+				stream = s
+			}
+
 			staleCount := 0
 			for _, spaceID := range spaceIDs {
-				staleCount += deleteStaleUploads(cfg, spaceID, dryRun, verbose)
+				staleCount += deleteStaleUploads(cfg, spaceID, dryRun, verbose, stream)
 			}
 
 			if verbose {
@@ -410,7 +421,7 @@ func globSpaceIDs(cfg *config.Config) []string {
 }
 
 // delete stale processing nodes for a given spaceID
-func deleteStaleUploads(cfg *config.Config, spaceID string, dryRun bool, verbose bool) int {
+func deleteStaleUploads(cfg *config.Config, spaceID string, dryRun bool, verbose bool, stream events.Stream) int {
 	if verbose {
 		fmt.Printf("\nDeleting stale processing nodes for space: %s\n", spaceID)
 	}
@@ -440,7 +451,7 @@ func deleteStaleUploads(cfg *config.Config, spaceID string, dryRun bool, verbose
 
 	staleCount := 0
 	for _, path := range mpkFiles {
-		staleCount += deleteStaleNode(cfg, path, dryRun, verbose)
+		staleCount += deleteStaleNode(cfg, path, dryRun, verbose, stream)
 	}
 
 	if verbose {
@@ -452,7 +463,7 @@ func deleteStaleUploads(cfg *config.Config, spaceID string, dryRun bool, verbose
 
 // deleteStaleNode deletes a stale node: if it is not referenced by any upload session
 // returns 1 if the node stale node was detected for deletion, 0 otherwise, for counting purposes
-func deleteStaleNode(cfg *config.Config, path string, dryRun bool, verbose bool) int {
+func deleteStaleNode(cfg *config.Config, path string, dryRun bool, verbose bool, stream events.Stream) int {
 	nodeDir := filepath.Dir(path)
 
 	// Read .mpk file to get processing info
@@ -495,11 +506,20 @@ func deleteStaleNode(cfg *config.Config, path string, dryRun bool, verbose bool)
 		return 1
 	}
 
-	if err := os.RemoveAll(nodeDir); err != nil {
-		fmt.Fprintf(os.Stderr, "%sError deleting stale node %s: %v\n", LOG_INDENT_L2, nodeDir, err)
+	rid := extractResourceID(strings.TrimSuffix(path, ".mpk"))
+	if rid == nil {
+		fmt.Fprintf(os.Stderr, "Failed to extract resource ID from path %s\n", path)
 		return 0
 	}
 
+	if err := events.Publish(context.Background(), stream, events.RevertRevision{
+		ResourceID: rid,
+		Timestamp:  utils.TSNow(),
+	}); err != nil {
+		// if publishing fails there is no need to try publishing other events - they will fail too.
+		log.Fatalf("Failed to send revert revision event for node '%s'\n", path)
+	}
+
 	if verbose {
 		fmt.Printf("%sDeleted stale node: %s\n", LOG_INDENT_L2, nodeDir)
 	}
@@ -518,3 +538,21 @@ func extractProcessingID(mpkData map[string]interface{}) string {
 	}
 	return processingID
 }
+
+func extractResourceID(path string) *provider.ResourceId {
+	// path looks like /.../storage/users/spaces/f2/06bccf-0f10-4070-9e63-40943f060667/nodes/5b/ba/1e/a7/-f185-4f31-8342-ed4b5743f096
+	parts := strings.Split(path, "spaces")
+	if len(parts) < 2 {
+		return nil
+	}
+
+	spaceParts := strings.Split(parts[1], "nodes")
+	if len(spaceParts) < 2 {
+		return nil
+	}
+
+	return &provider.ResourceId{
+		SpaceId:  strings.ReplaceAll(spaceParts[0], "/", ""),
+		OpaqueId: strings.ReplaceAll(spaceParts[1], "/", ""),
+	}
+}