feat(rst): Add priority and wait queue to sync.

Add priority option to ctl push/pull commands with five priority levels.
Add IsWorkRequestReady() method to the rst.Provider interface to validate work is ready.
* Before work is processed, IsWorkRequestReady() is called to determine if it should be added to the wait queue.

Fix invalid submission id parsing which resulted in each pullInNewWork() call to return "".

Author: swartzn

common/rst/builder.go

@@ -38,6 +38,7 @@ func (c *JobBuilderClient) GetJobRequest(cfg *flex.JobRequestCfg) *beeremote.Job
 		Path:                cfg.Path,
 		RemoteStorageTarget: 0,
 		StubLocal:           cfg.StubLocal,
+		Priority:            cfg.Priority,
 		Force:               cfg.Force,
 		Type: &beeremote.JobRequest_Builder{
 			Builder: &flex.BuilderJob{
@@ -92,6 +93,10 @@ func (c *JobBuilderClient) ExecuteJobBuilderRequest(ctx context.Context, workReq
 	return c.executeJobBuilderRequest(ctx, workRequest, walkChan, jobSubmissionChan, cfg)
 }
 
+func (r *JobBuilderClient) IsWorkRequestReady(ctx context.Context, request *flex.WorkRequest) (bool, time.Time, error) {
+	return true, time.Time{}, nil
+}
+
 // ExecuteWorkRequestPart is not implemented and should never be called.
 func (c *JobBuilderClient) ExecuteWorkRequestPart(ctx context.Context, request *flex.WorkRequest, part *flex.Work_Part) error {
 	return ErrUnsupportedOpForRST

common/rst/mock.go

@@ -126,3 +126,7 @@ func (r *MockClient) GetRemotePathInfo(ctx context.Context, cfg *flex.JobRequest
 func (r *MockClient) GenerateExternalId(ctx context.Context, cfg *flex.JobRequestCfg) (string, error) {
 	return "", ErrUnsupportedOpForRST
 }
+
+func (r *MockClient) IsWorkRequestReady(ctx context.Context, request *flex.WorkRequest) (bool, time.Time, error) {
+	return true, time.Time{}, nil
+}

common/rst/rst.go

@@ -110,6 +110,10 @@ type Provider interface {
 	GetRemotePathInfo(ctx context.Context, cfg *flex.JobRequestCfg) (remoteSize int64, remoteMtime time.Time, err error)
 	// GenerateExternalId can be used to generate an identifier for remote operations.
 	GenerateExternalId(ctx context.Context, cfg *flex.JobRequestCfg) (externalId string, err error)
+	// IsWorkRequestReady is used to indicate when the work request is ready and will be used to
+	// start work requests that have been placed into a wait queue. This is useful for providers
+	// that need the ability to wait for resources to be made available before continuing.
+	IsWorkRequestReady(ctx context.Context, request *flex.WorkRequest) (ready bool, retryTime time.Time, err error)
 }
 
 // New initializes a provider client based on the provided config. It accepts a context that can be

common/rst/s3.go

@@ -91,6 +91,7 @@ func (r *S3Client) GetJobRequest(cfg *flex.JobRequestCfg) *beeremote.JobRequest
 		Path:                cfg.Path,
 		RemoteStorageTarget: cfg.RemoteStorageTarget,
 		StubLocal:           cfg.StubLocal,
+		Priority:            cfg.Priority,
 		Force:               cfg.Force,
 		Type: &beeremote.JobRequest_Sync{
 			Sync: &flex.SyncJob{
@@ -114,6 +115,7 @@ func (r *S3Client) getJobRequestCfg(request *beeremote.JobRequest) *flex.JobRequ
 		StubLocal:           request.StubLocal,
 		Overwrite:           sync.Overwrite,
 		Flatten:             sync.Flatten,
+		Priority:            request.Priority,
 		Force:               request.Force,
 		LockedInfo:          sync.LockedInfo,
 	}
@@ -180,6 +182,10 @@ func (r *S3Client) ExecuteJobBuilderRequest(ctx context.Context, workRequest *fl
 	return ErrUnsupportedOpForRST
 }
 
+func (r *S3Client) IsWorkRequestReady(ctx context.Context, request *flex.WorkRequest) (bool, time.Time, error) {
+	return true, time.Time{}, nil
+}
+
 func (r *S3Client) ExecuteWorkRequestPart(ctx context.Context, request *flex.WorkRequest, part *flex.Work_Part) error {
 	if !request.HasSync() {
 		return ErrReqAndRSTTypeMismatch

ctl/internal/cmd/rst/pushpull.go

@@ -42,6 +42,10 @@ WARNING: Files are always uploaded and existing files overwritten unless the rem
 			if len(args) > 1 {
 				return fmt.Errorf("invalid number of arguments. Be sure to quote file glob pattern")
 			}
+			if backendCfg.Priority < 1 || backendCfg.Priority > 5 {
+				return fmt.Errorf("invalid priority! Priority must be between 1 and 5")
+			}
+			backendCfg.Priority--
 			return nil
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
@@ -50,6 +54,7 @@ WARNING: Files are always uploaded and existing files overwritten unless the rem
 		},
 	}
 	cmd.Flags().Uint32VarP(&backendCfg.RemoteStorageTarget, "remote-target", "r", 0, "Perform a one time push to the specified Remote Storage Target ID.")
+	cmd.Flags().Int32Var(&backendCfg.Priority, "priority", 3, "Job priority level (1-5; 1=lowest, 5=highest)")
 	cmd.Flags().BoolVar(&backendCfg.Force, "force", false, "Force push file(s) to the remote target even if the file is already in sync or another client currently has them open for writing (note the job may later fail or the uploaded file may not be the latest version).")
 	cmd.Flags().MarkHidden("force")
 	cmd.Flags().BoolVarP(&frontendCfg.verbose, "verbose", "v", false, "Print additional details about each job (use --debug) to also print work requests and results.")
@@ -70,6 +75,10 @@ func newPullCmd() *cobra.Command {
 			if len(args) != 1 {
 				return fmt.Errorf("missing <path> argument")
 			}
+			if backendCfg.Priority < 1 || backendCfg.Priority > 5 {
+				return fmt.Errorf("invalid priority! Priority must be between 1 and 5")
+			}
+			backendCfg.Priority--
 			return nil
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
@@ -83,6 +92,7 @@ func newPullCmd() *cobra.Command {
 	cmd.Flags().BoolVarP(&backendCfg.StubLocal, "stub-local", "s", false, "Create stub files for the remote objects or files.")
 	cmd.Flags().BoolVar(&backendCfg.Flatten, "flatten", false, "Flatten the remote directory structure. The directory delimiter will be replaced with an underscore.")
 	cmd.Flags().BoolVar(&backendCfg.Force, "force", false, "Force pulling file(s) from the remote target even if the file is already in sync or another client currently has them open for reading or writing (note other clients may see errors, the job may later fail, or the downloaded file may not be the latest version).")
+	cmd.Flags().Int32Var(&backendCfg.Priority, "priority", 3, "Job priority level (1-5; 1=lowest, 5=highest)")
 	cmd.Flags().MarkHidden("force")
 	cmd.Flags().BoolVarP(&frontendCfg.verbose, "verbose", "v", false, "Print additional details about each job (use --debug) to also print work requests and results.")
 	cmd.Flags().IntVar(&frontendCfg.width, "column-width", 35, "Set the maximum width of some columns before they overflow.")