Review comments

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

pkg/scheduler/exportimport.go

@@ -118,7 +118,6 @@ func Import(ctx context.Context, opts ExportImportOptions) error {
 	ops := make([]clientv3.Op, 0, len(in.Jobs)+len(in.Counters))
 
 	for key, b := range in.Jobs {
-		// Optional: verify bytes are valid before writing
 		var j stored.Job
 		if err := proto.Unmarshal(b, &j); err != nil {
 			return fmt.Errorf("unmarshal job %q: %w", key, err)
@@ -134,19 +133,21 @@ func Import(ctx context.Context, opts ExportImportOptions) error {
 		ops = append(ops, clientv3.OpPut(key, string(b)))
 	}
 
+	var end int
 	for i := 0; i < len(ops); i += 128 {
 		txn := client.Txn(ctx)
-		end := i + 128
+		end = i + 128
 		if end > len(ops) {
 			end = len(ops)
 		}
 		txn.Then(ops[i:end]...)
 		if _, err := txn.Commit(); err != nil {
+			print.FailureStatusEvent(os.Stderr, "Incomplete import with %d items.", end)
 			return fmt.Errorf("commit transaction: %w", err)
 		}
-
-		print.InfoStatusEvent(os.Stdout, "Imported %d items.", end)
 	}
 
+	print.InfoStatusEvent(os.Stdout, "Imported %d items.", end)
+
 	return nil
 }

pkg/scheduler/get.go

@@ -99,34 +99,37 @@ func getSingle(ctx context.Context, cl *clientv3.Client, key string, opts GetOpt
 }
 
 func pathsFromJobKey(jobKey *jobKey, namespace string) [2]string {
+	const reminderPath = "dapr/jobs/actorreminder"
+	const reminderCounterPath = "dapr/counters/actorreminder"
+
 	var paths [2]string
 	switch {
 	case jobKey.actorType != nil:
-		paths[0] = fmt.Sprintf("dapr/jobs/actorreminder||%s||%s||%s||%s",
-			namespace, *jobKey.actorType, *jobKey.actorID, jobKey.name,
+		paths[0] = fmt.Sprintf("%s||%s||%s||%s||%s",
+			reminderPath, namespace, *jobKey.actorType, *jobKey.actorID, jobKey.name,
 		)
-		paths[1] = fmt.Sprintf("dapr/counters/actorreminder||%s||%s||%s||%s",
-			namespace, *jobKey.actorType, *jobKey.actorID, jobKey.name,
+		paths[1] = fmt.Sprintf("%s||%s||%s||%s||%s",
+			reminderCounterPath, namespace, *jobKey.actorType, *jobKey.actorID, jobKey.name,
 		)
 
 	case jobKey.activity:
 		actorType := fmt.Sprintf("dapr.internal.%s.%s.activity", namespace, *jobKey.appID)
 		actorID := jobKey.name
-		paths[0] = fmt.Sprintf("dapr/jobs/actorreminder||%s||%s||%s||run-activity",
-			namespace, actorType, actorID,
+		paths[0] = fmt.Sprintf("%s||%s||%s||%s||run-activity",
+			reminderPath, namespace, actorType, actorID,
 		)
-		paths[1] = fmt.Sprintf("dapr/counters/actorreminder||%s||%s||%s||run-activity",
-			namespace, actorType, actorID,
+		paths[1] = fmt.Sprintf("%s||%s||%s||%s||run-activity",
+			reminderCounterPath, namespace, actorType, actorID,
 		)
 
 	case jobKey.instanceID != nil:
 		actorType := fmt.Sprintf("dapr.internal.%s.%s.workflow", namespace, *jobKey.appID)
 		actorID := *jobKey.instanceID
-		paths[0] = fmt.Sprintf("dapr/jobs/actorreminder||%s||%s||%s||%s",
-			namespace, actorType, actorID, jobKey.name,
+		paths[0] = fmt.Sprintf("%s||%s||%s||%s||%s",
+			reminderPath, namespace, actorType, actorID, jobKey.name,
 		)
-		paths[1] = fmt.Sprintf("dapr/counters/actorreminder||%s||%s||%s||%s",
-			namespace, actorType, actorID, jobKey.name,
+		paths[1] = fmt.Sprintf("%s||%s||%s||%s||%s",
+			reminderCounterPath, namespace, actorType, actorID, jobKey.name,
 		)
 
 	default:

pkg/scheduler/scheduler.go

@@ -148,7 +148,7 @@ func parseJobKey(key string) (*jobKey, error) {
 	}
 
 	switch split[0] {
-	case "app":
+	case FilterApp:
 		if len(split) != 3 {
 			return nil, fmt.Errorf("expecting job key to be in format 'app/{app ID}/{job name}', got '%s'", key)
 		}
@@ -157,7 +157,7 @@ func parseJobKey(key string) (*jobKey, error) {
 			name:  split[2],
 		}, nil
 
-	case "actor":
+	case FilterActor:
 		if len(split) != 4 {
 			return nil, fmt.Errorf("expecting actor reminder key to be in format 'actor/{actor type}/{actor id}/{name}', got '%s'", key)
 		}
@@ -167,19 +167,19 @@ func parseJobKey(key string) (*jobKey, error) {
 			name:      split[3],
 		}, nil
 
-	case "workflow":
+	case FilterWorkflow:
 		if len(split) != 4 {
-			return nil, fmt.Errorf("expecting worklow key to be in format 'workflow/{app ID}/{instance ID}/{name}', got '%s'", key)
+			return nil, fmt.Errorf("expecting workflow key to be in format 'workflow/{app ID}/{instance ID}/{name}', got '%s'", key)
 		}
 		return &jobKey{
 			appID:      &split[1],
 			instanceID: &split[2],
 			name:       split[3],
 		}, nil
 
-	case "activity":
+	case FilterActivity:
 		if len(split) != 3 {
-			return nil, fmt.Errorf("expecting activity key to be in format 'activity/{app ID}/{activity ID}/', got '%s'", key)
+			return nil, fmt.Errorf("expecting activity key to be in format 'activity/{app ID}/{activity ID}', got '%s'", key)
 		}
 		return &jobKey{
 			appID:    &split[1],
@@ -188,7 +188,7 @@ func parseJobKey(key string) (*jobKey, error) {
 		}, nil
 
 	default:
-		return nil, fmt.Errorf("unsupported job type '%s', accepts 'app', 'actor', or 'workflow'", split[0])
+		return nil, fmt.Errorf("unsupported job type '%s', accepts 'app', 'actor', 'workflow', or 'activity'", split[0])
 	}
 }