[full-ci] Postprocessing Bulk restart

changelog/unreleased/bump-reva.md

@@ -2,6 +2,7 @@ Enhancement: Update reva to latest edge version
 
 We update reva to the latest edge version to get the latest fixes and features.
 
+https://github.com/owncloud/ocis/pull/8287
 https://github.com/owncloud/ocis/pull/8278
 https://github.com/owncloud/ocis/pull/8264
 https://github.com/owncloud/ocis/pull/8100

changelog/unreleased/postprocessing-bulk-restart.md

@@ -0,0 +1,5 @@
+Enhancement: Allow restarting multiple uploads with one command
+
+Allows to restart all commands in a specific state.
+
+https://github.com/owncloud/ocis/pull/8287

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/coreos/go-oidc v2.2.1+incompatible
 	github.com/coreos/go-oidc/v3 v3.9.0
 	github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781
-	github.com/cs3org/reva/v2 v2.18.1-0.20240124094635-6eec406c0be7
+	github.com/cs3org/reva/v2 v2.18.1-0.20240126141248-c9e4a3bcd0da
 	github.com/dhowden/tag v0.0.0-20230630033851-978a0926ee25
 	github.com/disintegration/imaging v1.6.2
 	github.com/dutchcoders/go-clamd v0.0.0-20170520113014-b970184f4d9e

go.sum

@@ -1018,8 +1018,8 @@ github.com/crewjam/saml v0.4.14 h1:g9FBNx62osKusnFzs3QTN5L9CVA/Egfgm+stJShzw/c=
 github.com/crewjam/saml v0.4.14/go.mod h1:UVSZCf18jJkk6GpWNVqcyQJMD5HsRugBPf4I1nl2mME=
 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781 h1:BUdwkIlf8IS2FasrrPg8gGPHQPOrQ18MS1Oew2tmGtY=
 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781/go.mod h1:UXha4TguuB52H14EMoSsCqDj7k8a/t7g4gVP+bgY5LY=
-github.com/cs3org/reva/v2 v2.18.1-0.20240124094635-6eec406c0be7 h1:g7vQAbo64ziFqqhKcim3JCjDW1zqHy9imAm2HZmmK8w=
-github.com/cs3org/reva/v2 v2.18.1-0.20240124094635-6eec406c0be7/go.mod h1:GCN3g6uYE0Nvd31dGlhaGGyUviUfbG2NkecPRv5oSc4=
+github.com/cs3org/reva/v2 v2.18.1-0.20240126141248-c9e4a3bcd0da h1:VgWIr/lE6cv2f5IjjWgR0LOAK41gsUytBsSZo/4DRq4=
+github.com/cs3org/reva/v2 v2.18.1-0.20240126141248-c9e4a3bcd0da/go.mod h1:GCN3g6uYE0Nvd31dGlhaGGyUviUfbG2NkecPRv5oSc4=
 github.com/cyberdelia/templates v0.0.0-20141128023046-ca7fffd4298c/go.mod h1:GyV+0YP4qX0UQ7r2MoYZ+AvYDp12OF5yg4q8rGnyNh4=
 github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg=
 github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4=

services/postprocessing/README.md

@@ -90,3 +90,11 @@ ocis storage-users uploads list
 ```bash
 ocis postprocessing restart -u <uploadID>
 ```
+
+Instead of starting one specific upload, a system admin can also restart all uploads that are currently in a specific step.
+Examples:
+```
+ocis postprocessing restart # Restarts all uploads where postprocessing is finished, but upload is not finished
+ocis postprocessing restart -s "finished" # Equivalent to the above
+ocis postprocessing restart -s "virusscan" # Restart all uploads currently in virusscan step
+```

services/postprocessing/pkg/command/postprocessing.go

@@ -2,7 +2,6 @@ package command
 
 import (
 	"context"
-	"time"
 
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/events/stream"
@@ -20,10 +19,15 @@ func RestartPostprocessing(cfg *config.Config) *cli.Command {
 		Usage: "restart postprocessing for an uploadID",
 		Flags: []cli.Flag{
 			&cli.StringFlag{
-				Name:     "upload-id",
-				Aliases:  []string{"u"},
-				Required: true,
-				Usage:    "the uploadid to restart",
+				Name:    "upload-id",
+				Aliases: []string{"u"},
+				Usage:   "the uploadid to restart. Ignored if unset.",
+			},
+			&cli.StringFlag{
+				Name:    "step",
+				Aliases: []string{"s"},
+				Usage:   "restarts all uploads in the given postprocessing step. Ignored if upload-id is set.",
+				Value:   "finished", // Calling `ocis postprocessing restart` without any arguments will restart all uploads that are finished but failed to move the uploed from the upload area to the blobstore.
 			},
 		},
 		Before: func(c *cli.Context) error {
@@ -35,24 +39,18 @@ func RestartPostprocessing(cfg *config.Config) *cli.Command {
 				return err
 			}
 
-			ev := events.ResumePostprocessing{
-				UploadID:  c.String("upload-id"),
-				Timestamp: utils.TSNow(),
+			uid, step := c.String("upload-id"), ""
+			if uid == "" {
+				step = c.String("step")
 			}
 
-			if err := events.Publish(context.Background(), stream, ev); err != nil {
-				return err
+			ev := events.ResumePostprocessing{
+				UploadID:  uid,
+				Step:      events.Postprocessingstep(step),
+				Timestamp: utils.TSNow(),
 			}
 
-			// go-micro nats implementation uses async publishing,
-			// therefore we need to manually wait.
-			//
-			// FIXME: upstream pr
-			//
-			// https://github.com/go-micro/plugins/blob/3e77393890683be4bacfb613bc5751867d584692/v4/events/natsjs/nats.go#L115
-			time.Sleep(5 * time.Second)
-
-			return nil
+			return events.Publish(context.Background(), stream, ev)
 		},
 	}
 }

services/postprocessing/pkg/postprocessing/postprocessing.go

@@ -65,14 +65,14 @@ func (pp *Postprocessing) NextStep(ev events.PostprocessingStepFinished) interfa
 
 // CurrentStep returns the current postprocessing step
 func (pp *Postprocessing) CurrentStep() interface{} {
-	if pp.Status.Outcome != "" {
+	if pp.Status.CurrentStep == events.PPStepFinished {
 		return pp.finished(pp.Status.Outcome)
 	}
 	return pp.step(pp.Status.CurrentStep)
 }
 
 // Delay will sleep the configured time then continue
-func (pp *Postprocessing) Delay(ev events.StartPostprocessingStep) interface{} {
+func (pp *Postprocessing) Delay() interface{} {
 	time.Sleep(pp.config.Delayprocessing)
 	return pp.next(events.PPStepDelay)
 }
@@ -106,6 +106,7 @@ func (pp *Postprocessing) step(next events.Postprocessingstep) events.StartPostp
 }
 
 func (pp *Postprocessing) finished(outcome events.PostprocessingOutcome) events.PostprocessingFinished {
+	pp.Status.CurrentStep = events.PPStepFinished
 	pp.Status.Outcome = outcome
 	return events.PostprocessingFinished{
 		UploadID:      pp.ID,

services/postprocessing/pkg/service/service.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/cs3org/reva/v2/pkg/events"
+	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/services/postprocessing/pkg/config"
 	"github.com/owncloud/ocis/v2/services/postprocessing/pkg/postprocessing"
@@ -142,29 +143,20 @@ func (pps *PostprocessingService) processEvent(e events.Event) error {
 			pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot get upload")
 			return fmt.Errorf("%w: cannot get upload", errEvent)
 		}
-		next = pp.Delay(ev)
+		next = pp.Delay()
 	case events.UploadReady:
+		if ev.Failed {
+			// the upload failed - let's keep it around for a while
+			return nil
+		}
+
 		// the storage provider thinks the upload is done - so no need to keep it any more
 		if err := pps.store.Delete(ev.UploadID); err != nil {
 			pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot delete upload")
 			return fmt.Errorf("%w: cannot delete upload", errEvent)
 		}
 	case events.ResumePostprocessing:
-		pp, err = pps.getPP(pps.store, ev.UploadID)
-		if err != nil {
-			if err == store.ErrNotFound {
-				if err := events.Publish(ctx, pps.pub, events.RestartPostprocessing{
-					UploadID:  ev.UploadID,
-					Timestamp: ev.Timestamp,
-				}); err != nil {
-					pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot publish RestartPostprocessing event")
-				}
-				return fmt.Errorf("%w: cannot publish RestartPostprocessing event", errEvent)
-			}
-			pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot get upload")
-			return fmt.Errorf("%w: cannot get upload", errEvent)
-		}
-		next = pp.CurrentStep()
+		return pps.handleResumePPEvent(ctx, ev)
 	}
 
 	if pp != nil {
@@ -182,11 +174,30 @@ func (pps *PostprocessingService) processEvent(e events.Event) error {
 	return nil
 }
 
+func (pps *PostprocessingService) getPP(sto store.Store, uploadID string) (*postprocessing.Postprocessing, error) {
+	recs, err := sto.Read(uploadID)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(recs) != 1 {
+		return nil, fmt.Errorf("expected only one result for '%s', got %d", uploadID, len(recs))
+	}
+
+	pp := postprocessing.New(pps.c)
+	err = json.Unmarshal(recs[0].Value, pp)
+	if err != nil {
+		return nil, err
+	}
+
+	return pp, nil
+}
+
 func getSteps(c config.Postprocessing) []events.Postprocessingstep {
 	// NOTE: improved version only allows configuring order of postprocessing steps
 	// But we aim for a system where postprocessing steps can be configured per space, ideally by the spaceadmin itself
 	// We need to iterate over configuring PP service when we see fit
-	var steps []events.Postprocessingstep
+	steps := make([]events.Postprocessingstep, 0, len(c.Steps))
 	for _, s := range c.Steps {
 		steps = append(steps, events.Postprocessingstep(s))
 	}
@@ -206,21 +217,70 @@ func storePP(sto store.Store, pp *postprocessing.Postprocessing) error {
 	})
 }
 
-func (pps *PostprocessingService) getPP(sto store.Store, uploadID string) (*postprocessing.Postprocessing, error) {
-	recs, err := sto.Read(uploadID)
-	if err != nil {
-		return nil, err
+func (pps *PostprocessingService) handleResumePPEvent(ctx context.Context, ev events.ResumePostprocessing) error {
+	ids := []string{ev.UploadID}
+	if ev.Step != "" {
+		ids = pps.findUploadsByStep(ev.Step)
 	}
 
-	if len(recs) != 1 {
-		return nil, fmt.Errorf("expected only one result for '%s', got %d", uploadID, len(recs))
+	for _, id := range ids {
+		if err := pps.resumePP(ctx, id); err != nil {
+			pps.log.Error().Str("uploadID", id).Err(err).Msg("cannot resume upload")
+			return fmt.Errorf("%w: cannot resume upload", errEvent)
+		}
 	}
+	return nil
+}
 
-	pp := postprocessing.New(pps.c)
-	err = json.Unmarshal(recs[0].Value, pp)
+func (pps *PostprocessingService) resumePP(ctx context.Context, uploadID string) error {
+	pp, err := pps.getPP(pps.store, uploadID)
 	if err != nil {
-		return nil, err
+		if err == store.ErrNotFound {
+			if err := events.Publish(ctx, pps.pub, events.RestartPostprocessing{
+				UploadID:  uploadID,
+				Timestamp: utils.TSNow(),
+			}); err != nil {
+				return err
+			}
+			return nil
+		}
+		return fmt.Errorf("%w: cannot get upload", errEvent)
 	}
 
-	return pp, nil
+	return events.Publish(ctx, pps.pub, pp.CurrentStep())
+}
+
+func (pps *PostprocessingService) findUploadsByStep(step events.Postprocessingstep) []string {
+	var ids []string
+
+	keys, err := pps.store.List()
+	if err != nil {
+		pps.log.Error().Err(err).Msg("cannot list uploads")
+	}
+
+	for _, k := range keys {
+		rec, err := pps.store.Read(k)
+		if err != nil {
+			pps.log.Error().Err(err).Msg("cannot read upload")
+			continue
+		}
+
+		if len(rec) != 1 {
+			pps.log.Error().Err(err).Msg("expected only one result")
+			continue
+		}
+
+		pp := &postprocessing.Postprocessing{}
+		err = json.Unmarshal(rec[0].Value, pp)
+		if err != nil {
+			pps.log.Error().Err(err).Msg("cannot unmarshal upload")
+			continue
+		}
+
+		if pp.Status.CurrentStep == step {
+			ids = append(ids, pp.ID)
+		}
+	}
+
+	return ids
 }

tests/acceptance/features/apiSpacesDavOperation/moveByFileId.feature

@@ -94,8 +94,8 @@ Feature: moving/renaming file using file id
     And user "Brian" has uploaded file with content "some data" to "/test.txt"
     And we save it into "FILEID"
     When user "Brian" moves a file "test.txt" into "folder" inside space "Shares" using file-id path "<dav-path>"
-    Then the HTTP status code should be "403"
-    And the value of the item "/d:error/s:message" in the response about user "Brian" should be "cross storage moves are not permitted, use copy and delete"
+    Then the HTTP status code should be "502"
+    And the value of the item "/d:error/s:message" in the response about user "Brian" should be "cross storage moves are not supported, use copy and delete"
     And for user "Brian" folder "/" of the space "Personal" should contain these files:
       | test.txt |
     But for user "Alice" folder "folder" of the space "Personal" should not contain these files:
@@ -351,7 +351,7 @@ Feature: moving/renaming file using file id
     And user "Alice" has created folder "testshare"
     And user "Alice" has shared folder "testshare" with user "Brian" with permissions "<permissions>"
     When user "Brian" moves a file "textfile.txt" into "testshare" inside space "Shares" using file-id path "<dav-path>"
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Brian" folder "/" of the space "project-space" should contain these files:
       | textfile.txt |
     But for user "Brian" folder "testshare" of the space "Shares" should not contain these files:
@@ -475,7 +475,7 @@ Feature: moving/renaming file using file id
     And we save it into "FILEID"
     And user "Alice" has shared folder "folder" with user "Brian" with permissions "read"
     When user "Brian" moves a file "Shares/folder/test.txt" into "folder/sub-folder" inside space "Shares" using file-id path "<dav-path>"
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Brian" folder "folder/sub-folder" of the space "Shares" should not contain these files:
       | test.txt |
     And for user "Alice" folder "folder/sub-folder" of the space "Personal" should not contain these files:
@@ -499,7 +499,7 @@ Feature: moving/renaming file using file id
     And user "Alice" has shared folder "testshare1" with user "Brian" with permissions "<from_permissions>"
     And user "Alice" has shared folder "testshare2" with user "Brian" with permissions "<to_permissions>"
     When user "Brian" moves a file "Shares/testshare1/textfile.txt" into "testshare2" inside space "Shares" using file-id path "<dav-path>"
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Brian" folder "testshare1" of the space "Shares" should contain these files:
       | textfile.txt |
     But for user "Brian" folder "testshare2" of the space "Shares" should not contain these files:
@@ -697,7 +697,7 @@ Feature: moving/renaming file using file id
     And we save it into "FILEID"
     And user "Alice" has shared folder "/folder" with user "Brian" with permissions "read"
     When user "Brian" renames a file "Shares/folder/test.txt" into "folder/sub-folder/renamed.txt" inside space "Shares" using file-id path "<dav-path>"
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Brian" folder "folder" of the space "Shares" should contain these files:
       | test.txt |
     But for user "Brian" folder "folder/sub-folder" of the space "Shares" should not contain these files:

tests/acceptance/features/apiSpacesShares/moveSpaces.feature

@@ -125,7 +125,7 @@ Feature: move (rename) file
       | role      | <role> |
     And user "Brian" has shared folder "/testshare" with user "Alice" with permissions "<permissions>"
     When user "Alice" moves file "project.txt" from space "Project" to "/testshare/project.txt" inside space "Shares" using the WebDAV API
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Alice" the space "Project" should contain these entries:
       | project.txt |
     But for user "Alice" folder "testshare" of the space "Shares" should not contain these entries:
@@ -168,7 +168,7 @@ Feature: move (rename) file
     And user "Brian" has shared folder "/testshare" with user "Alice" with permissions "<permissions>"
     And user "Alice" has uploaded file with content "personal content" to "personal.txt"
     When user "Alice" moves file "personal.txt" from space "Personal" to "/testshare/personal.txt" inside space "Shares" using the WebDAV API
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Alice" the space "Personal" should contain these entries:
       | personal.txt |
     But for user "Alice" folder "testshare" of the space "Shares" should not contain these entries:
@@ -185,7 +185,7 @@ Feature: move (rename) file
     And user "Brian" has uploaded file with content "testshare content" to "/testshare/testshare.txt"
     And user "Brian" has shared folder "/testshare" with user "Alice" with permissions "<permissions>"
     When user "Alice" moves file "/testshare/testshare.txt" from space "Shares" to "testshare.txt" inside space "Personal" using the WebDAV API
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Alice" the space "Personal" should not contain these entries:
       | testshare.txt |
     And for user "Alice" folder "testshare" of the space "Shares" should contain these entries:
@@ -207,7 +207,7 @@ Feature: move (rename) file
     And user "Brian" has uploaded file with content "testshare content" to "/testshare/testshare.txt"
     And user "Brian" has shared folder "/testshare" with user "Alice" with permissions "<permissions>"
     When user "Alice" moves file "/testshare/testshare.txt" from space "Shares" to "testshare.txt" inside space "Project" using the WebDAV API
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Alice" the space "Project" should not contain these entries:
       | /testshare.txt |
     And for user "Alice" folder "testshare" of the space "Shares" should contain these entries:
@@ -232,7 +232,7 @@ Feature: move (rename) file
     And user "Brian" has shared folder "/testshare1" with user "Alice" with permissions "<from_permissions>"
     And user "Brian" has shared folder "/testshare2" with user "Alice" with permissions "<to_permissions>"
     When user "Alice" moves file "/testshare1/testshare1.txt" from space "Shares" to "/testshare2/testshare1.txt" inside space "Shares" using the WebDAV API
-    Then the HTTP status code should be "403"
+    Then the HTTP status code should be "502"
     And for user "Alice" folder "testshare1" of the space "Shares" should contain these entries:
       | testshare1.txt |
     But for user "Alice" folder "testshare2" of the space "Shares" should not contain these entries: