port the upload coordinator

go.mod

@@ -68,7 +68,7 @@ require (
 	github.com/open-policy-agent/opa v1.21.1
 	github.com/opencloud-eu/icap-client v0.0.0-20250930132611-28a2afe62d89
 	github.com/opencloud-eu/libre-graph-api-go v1.0.9-0.20261016073229-17ed3d5dd346
-	github.com/opencloud-eu/reva/v2 v2.52.1
+	github.com/opencloud-eu/reva/v2 v2.52.2-0.20261030103702-41d935e7864e
 	github.com/opensearch-project/opensearch-go/v4 v4.7.3
 	github.com/orcaman/concurrent-map v1.0.0
 	github.com/pkg/errors v0.9.1
@@ -218,7 +218,7 @@ require (
 	github.com/go-micro/plugins/v4/store/nats-js v1.2.1 // indirect
 	github.com/go-micro/plugins/v4/store/redis v1.2.1 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
-	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/locales v0.14.2 // indirect
 	github.com/go-playground/universal-translator v0.18.2 // indirect
 	github.com/go-redis/redis/v8 v8.11.5 // indirect
 	github.com/go-sql-driver/mysql v1.10.1 // indirect

go.sum

@@ -373,8 +373,8 @@ github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
 github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
 github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
-github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
-github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
+github.com/go-playground/locales v0.14.2 h1:d8UmcrM6Nip0hfGZKLGpAvZH37XB4TS0xzK9B56YNCY=
+github.com/go-playground/locales v0.14.2/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
 github.com/go-playground/universal-translator v0.18.2 h1:LCsMLC9RzmbUMNUPVYD15dmcjwYAJhmX8mPZRW4rAVU=
 github.com/go-playground/universal-translator v0.18.2/go.mod h1:67VZIMp5lQpDWlnStOct22q1bkdJGJqHghbOtmkawxk=
 github.com/go-playground/validator/v10 v10.30.3 h1:4MU6YkEwx7GbcPJOZxrtbu+QfF3pJLJuaYTeAH0DYy8=
@@ -761,8 +761,8 @@ github.com/opencloud-eu/icap-client v0.0.0-20250930132611-28a2afe62d89 h1:W1ms+l
 github.com/opencloud-eu/icap-client v0.0.0-20250930132611-28a2afe62d89/go.mod h1:vigJkNss1N2QEceCuNw/ullDehncuJNFB6mEnzfq9UI=
 github.com/opencloud-eu/libre-graph-api-go v1.0.9-0.20261016073229-17ed3d5dd346 h1:3ptbiFhSK4fzY/rC3idcYUveiwepMoOov1sBpc7Rg3k=
 github.com/opencloud-eu/libre-graph-api-go v1.0.9-0.20261016073229-17ed3d5dd346/go.mod h1:lTM8JeGblNpoMySTW7Lui2+c5TTLI95mwxtdUIHHrhU=
-github.com/opencloud-eu/reva/v2 v2.52.1 h1:c4tjmqtbJeqDjztTNq4Jvl3dWWSMHhXqV1JixKLKbT4=
-github.com/opencloud-eu/reva/v2 v2.52.1/go.mod h1:rAwuI3PX8n3RxBXHfpMtAKqfwxTIKhZol1Bi3Qj39p8=
+github.com/opencloud-eu/reva/v2 v2.52.2-0.20261030103702-41d935e7864e h1:uuUmKtZB9uRRVwqxYtZspEEQU0+Iacn1qAIuVek86DU=
+github.com/opencloud-eu/reva/v2 v2.52.2-0.20261030103702-41d935e7864e/go.mod h1:UNBaBcVdOojBaNYjcxJzqvvqVtk8695IAXkd3q6XPl8=
 github.com/opencloud-eu/secure v0.0.0-20260312082735-b6f5cb2244e4 h1:l2oB/RctH+t8r7QBj5p8thfEHCM/jF35aAY3WQ3hADI=
 github.com/opencloud-eu/secure v0.0.0-20260312082735-b6f5cb2244e4/go.mod h1:BmF5hyM6tXczk3MpQkFf1hpKSRqCyhqcbiQtiAF7+40=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=

services/ocm/pkg/revaconfig/config.go

@@ -3,6 +3,7 @@ package revaconfig
 import (
 	"math"
 	"net/url"
+	"path/filepath"
 
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/ocm/pkg/config"
@@ -94,8 +95,9 @@ func OCMConfigFromStruct(cfg *config.Config, logger log.Logger) map[string]any {
 					"expose_recipient_display_name": cfg.OCMD.ExposeRecipientDisplayName,
 				},
 				"dataprovider": map[string]any{
-					"prefix": "data",
-					"driver": "ocmreceived",
+					"prefix":           "data",
+					"driver":           "ocmreceived",
+					"upload_directory": filepath.Join(cfg.OCMStorageProvider.StorageRoot, "uploads"),
 					"drivers": map[string]any{
 						"ocmreceived": map[string]any{
 							"insecure":               cfg.OCMStorageProvider.Insecure,
@@ -175,7 +177,8 @@ func OCMConfigFromStruct(cfg *config.Config, logger log.Logger) map[string]any {
 					},
 				},
 				"storageprovider": map[string]any{
-					"driver": "ocmreceived",
+					"driver":           "ocmreceived",
+					"upload_directory": filepath.Join(cfg.OCMStorageProvider.StorageRoot, "uploads"),
 					"drivers": map[string]any{
 						"ocmreceived": map[string]any{
 							"insecure":     cfg.OCMStorageProvider.Insecure,

services/postprocessing/pkg/postprocessing/postprocessing.go

@@ -130,6 +130,7 @@ func (pp *Postprocessing) finished(outcome events.PostprocessingOutcome) events.
 		UploadID:          pp.ID,
 		ExecutingUser:     pp.User,
 		Filename:          pp.Filename,
+		ResourceID:        pp.ResourceID,
 		Outcome:           outcome,
 		ImpersonatingUser: pp.ImpersonatingUser,
 	}

services/storage-users/README.md

@@ -87,6 +87,7 @@ OPTIONS:
    --processing  filter sessions by processing status (default: unset)
    --expired     filter sessions by expired status (default: unset)
    --has-virus   filter sessions by virus scan result (default: unset)
+   --orphaned    filter sessions by whether their node metadata is unreadable (default: unset)
    --json        output as json (default: false)
    --restart     send restart event for all listed sessions (default: false)
    --resume      send resume event for all listed sessions (default: false)
@@ -141,6 +142,23 @@ opencloud storage-users uploads sessions --expired=true --clean
 opencloud storage-users uploads sessions --processing=false --has-virus=false --resume
 ```
 
+Uploads whose node metadata can no longer be read are orphaned: they can never finish
+postprocessing, because the destination of the upload cannot be resolved. Such uploads stay
+in processing state indefinitely and keep consuming the quota of their space. Use the
+`--orphaned` filter to list them, and `--clean` to remove them and release the quota.
+
+```bash
+# lists all orphaned upload sessions
+opencloud storage-users uploads sessions --orphaned=true
+
+# removes them and releases the quota they consume
+opencloud storage-users uploads sessions --orphaned=true --clean
+```
+
+Note: `--orphaned` reads the node metadata of every upload session, so it is slower than the
+other filters. Cleaning an orphaned session deletes the uploaded bytes, which are the only
+copy as long as postprocessing has not finished. Run the command without `--clean` first.
+
 
 #### Delete Stale Nodes command
 

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

@@ -20,6 +20,7 @@ import (
 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
+	olog "github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/storage-users/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/storage-users/pkg/config/parser"
 	"github.com/opencloud-eu/opencloud/services/storage-users/pkg/event"
@@ -30,6 +31,7 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/lookup"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/metadata/prefixes"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/node"
+	"github.com/opencloud-eu/reva/v2/pkg/upload"
 	"github.com/opencloud-eu/reva/v2/pkg/utils"
 )
 
@@ -92,10 +94,12 @@ func ListUploadSessions(cfg *config.Config) *cobra.Command {
 				// We need to init the posix driver with 'scanfs' disabled
 				posixDrivers := revaconfig.Posix(cfg, false, false)
 				// Disable async file uploads: this command only lists sessions and publishes
-				// events, but with async uploads enabled the driver would join the "dcfs"
-				// JetStream queue group. Events routed to this short-lived CLI member are
-				// auto-acked by the nats client and lost when the process exits, so the
-				// clean/restart/resume events published below would never reach the daemon.
+				// events, but with async uploads enabled the driver would join the
+				// "dcfs-revisions" JetStream queue group. Events routed to this short-lived
+				// CLI member are auto-acked by the nats client and lost when the process
+				// exits, so the daemon would miss them. The coordinator built below does not
+				// subscribe to postprocessing events, so the clean/restart/resume events
+				// published below reach the daemon.
 				posixDrivers["asyncfileuploads"] = false
 				drivers["posix"] = posixDrivers
 				// Also posix refuses to start without an events stream
@@ -106,15 +110,21 @@ func ListUploadSessions(cfg *config.Config) *cobra.Command {
 				}
 			}
 
-			fs, err := f(drivers[cfg.Driver].(map[string]any), fsStream, nil)
+			driverConf, ok := drivers[cfg.Driver].(map[string]any)
+			if !ok {
+				fmt.Fprintf(os.Stderr, "No configuration found for filesystem driver '%s'\n", cfg.Driver)
+				os.Exit(1)
+			}
+			fs, err := f(driverConf, fsStream, nil)
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "Failed to initialize filesystem driver '%s'\n", cfg.Driver)
 				return err
 			}
 
-			managingFS, ok := fs.(storage.UploadSessionLister)
-			if !ok {
-				fmt.Fprintf(os.Stderr, "'%s' storage does not support listing upload sessions\n", cfg.Driver)
+			logger := olog.Configure(cfg.Service.Name, cfg.Commons, cfg.LogLevel)
+			coord, err := upload.NewCoordinatorFromConfig("", driverConf, fs, nil, &logger.Logger, false)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Failed to initialize the upload coordinator: %v\n", err)
 				os.Exit(1)
 			}
 
@@ -133,7 +143,7 @@ func ListUploadSessions(cfg *config.Config) *cobra.Command {
 			}
 
 			filter := buildFilter(cmd)
-			uploads, err := managingFS.ListUploadSessions(cmd.Context(), filter)
+			uploads, err := coord.ListUploadSessions(cmd.Context(), filter)
 			if err != nil {
 				return err
 			}
@@ -238,6 +248,7 @@ func ListUploadSessions(cfg *config.Config) *cobra.Command {
 	listUploadSessionsCmd.Flags().Bool("processing", false, "filter sessions by processing status")
 	listUploadSessionsCmd.Flags().Bool("expired", false, "filter sessions by expired status")
 	listUploadSessionsCmd.Flags().Bool("has-virus", false, "filter sessions by virus scan result")
+	listUploadSessionsCmd.Flags().Bool("orphaned", false, "filter sessions by whether their node metadata is unreadable. Such sessions can never finish processing and can be removed with --clean")
 	listUploadSessionsCmd.Flags().Bool("json", false, "output as json")
 	listUploadSessionsCmd.Flags().Bool("restart", false, "send restart event for all listed sessions. Only one of resume/restart/clean can be set.")
 	listUploadSessionsCmd.Flags().Bool("resume", false, "send resume event for all listed sessions. Only one of resume/restart/clean can be set.")
@@ -259,6 +270,10 @@ func buildFilter(cmd *cobra.Command) storage.UploadSessionFilter {
 		infectedValue, _ := cmd.Flags().GetBool("has-virus")
 		filter.HasVirus = &infectedValue
 	}
+	if cmd.Flag("orphaned").Changed {
+		orphanedValue, _ := cmd.Flags().GetBool("orphaned")
+		filter.Orphaned = &orphanedValue
+	}
 	if cmd.Flag("id").Changed {
 		idValue, _ := cmd.Flags().GetString("id")
 		if idValue != "" {
@@ -317,6 +332,24 @@ func buildInfo(filter storage.UploadSessionFilter) string {
 		}
 	}
 
+	if filter.Orphaned != nil {
+		if b.Len() != 0 {
+			b.WriteString(", ")
+		}
+		if !*filter.Orphaned {
+			if b.Len() == 0 {
+				b.WriteString("Not ")
+			} else {
+				b.WriteString("not ")
+			}
+		}
+		if b.Len() == 0 {
+			b.WriteString("Orphaned")
+		} else {
+			b.WriteString("orphaned")
+		}
+	}
+
 	if b.Len() == 0 {
 		b.WriteString("Session")
 	} else {

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

@@ -68,6 +68,26 @@ func TestBuildInfo(t *testing.T) {
 			filter:       storage.UploadSessionFilter{Processing: boolPtr(true), Expired: boolPtr(false), HasVirus: boolPtr(true), ID: strPtr("123")},
 			expectedInfo: "Processing, not expired, virusinfected session with id '123':",
 		},
+		{
+			alias:        "orphaned",
+			filter:       storage.UploadSessionFilter{Orphaned: boolPtr(true)},
+			expectedInfo: "Orphaned sessions:",
+		},
+		{
+			alias:        "not orphaned",
+			filter:       storage.UploadSessionFilter{Orphaned: boolPtr(false)},
+			expectedInfo: "Not orphaned sessions:",
+		},
+		{
+			alias:        "processing and orphaned",
+			filter:       storage.UploadSessionFilter{Processing: boolPtr(true), Orphaned: boolPtr(true)},
+			expectedInfo: "Processing, orphaned sessions:",
+		},
+		{
+			alias:        "expired, not virus infected and orphaned",
+			filter:       storage.UploadSessionFilter{Expired: boolPtr(true), HasVirus: boolPtr(false), Orphaned: boolPtr(true)},
+			expectedInfo: "Expired, not virusinfected, orphaned sessions:",
+		},
 	}
 
 	for _, tc := range testCases {

services/storage-users/pkg/revaconfig/config.go

@@ -81,6 +81,7 @@ func StorageUsersConfigFromStruct(cfg *config.Config) map[string]any {
 					"prefix":                 cfg.HTTP.Prefix,
 					"driver":                 cfg.Driver,
 					"drivers":                DataProviderDrivers(cfg),
+					"mount_id":               cfg.MountID,
 					"nats_address":           cfg.Events.Addr,
 					"nats_clusterID":         cfg.Events.ClusterID,
 					"nats_tls_insecure":      cfg.Events.TLSInsecure,

services/storage-users/pkg/revaconfig/drivers.go

@@ -87,6 +87,7 @@ func Local(cfg *config.Config) map[string]any {
 // Posix is the config mapping for the Posix storage driver
 func Posix(cfg *config.Config, enableFSScan, enableFSWatch bool) map[string]any {
 	return map[string]any{
+		"mount_id":                    cfg.MountID,
 		"root":                        cfg.Drivers.Posix.Root,
 		"personalspacepath_template":  cfg.Drivers.Posix.PersonalSpacePathTemplate,
 		"personalspacealias_template": cfg.Drivers.Posix.PersonalSpaceAliasTemplate,
@@ -183,6 +184,7 @@ func OwnCloudSQL(cfg *config.Config) map[string]any {
 // Decomposed is the config mapping for the Decomposed storage driver
 func Decomposed(cfg *config.Config) map[string]any {
 	return map[string]any{
+		"mount_id":         cfg.MountID,
 		"metadata_backend": "messagepack",
 		"metadata_prefix":  cfg.Drivers.Decomposed.MetadataPrefix,
 		"propagator":       cfg.Drivers.Decomposed.Propagator,
@@ -243,6 +245,7 @@ func Decomposed(cfg *config.Config) map[string]any {
 // DecomposedsNoEvents is the config mapping for the Decomposed storage driver emitting no events
 func DecomposedNoEvents(cfg *config.Config) map[string]any {
 	return map[string]any{
+		"mount_id":         cfg.MountID,
 		"metadata_backend": "messagepack",
 		"metadata_prefix":  cfg.Drivers.Decomposed.MetadataPrefix,
 		"propagator":       cfg.Drivers.Decomposed.Propagator,
@@ -296,6 +299,7 @@ func DecomposedNoEvents(cfg *config.Config) map[string]any {
 // DecomposedS3 is the config mapping for the decomposeds3 storage driver
 func DecomposedS3(cfg *config.Config) map[string]any {
 	return map[string]any{
+		"mount_id":         cfg.MountID,
 		"metadata_backend": "messagepack",
 		"metadata_prefix":  cfg.Drivers.DecomposedS3.MetadataPrefix,
 		"propagator":       cfg.Drivers.DecomposedS3.Propagator,
@@ -366,6 +370,7 @@ func DecomposedS3(cfg *config.Config) map[string]any {
 // DecomposedS3NoEvents is the config mapping for the decomposeds3 storage driver emitting no events
 func DecomposedS3NoEvents(cfg *config.Config) map[string]any {
 	return map[string]any{
+		"mount_id":         cfg.MountID,
 		"metadata_backend": "messagepack",
 		"metadata_prefix":  cfg.Drivers.DecomposedS3.MetadataPrefix,
 		"propagator":       cfg.Drivers.DecomposedS3.Propagator,

tests/acceptance/features/apiOcm/share.feature

@@ -1136,7 +1136,7 @@ Feature: an user shares resources using ScienceMesh application
       | resource           | textfile.txt             |
     And using server "REMOTE"
     When user "Brian" updates the content of federated share "textfile.txt" with "this is a new content" using the WebDAV API
-    Then the HTTP status code should be "500"
+    Then the HTTP status code should be "403"
     And using server "LOCAL"
     And the content of file "textfile.txt" for user "Alice" should be "ocm test"