修复 spark_submit 参数重复、上传校验绕过等 8 处缺陷

对应代码审查发现的问题（#1, #2, #4-#8, #10）：

#1 CRITICAL：spark_submit 在 argv 中重复拼接 binary。此前
    cmd := []string{binary} 后又把 cmd 作为 Args 传给 executor.Run，
    而 executor 会再拼一次 Binary，导致 OS argv 为 [binary, binary, ...]，
    spark-submit 会把自身当作应用 jar。现在 cmd 从 --master 开始，
    executor.Run 使用 Binary + Args，argv 正确。
#2：Validate 曾接受 .meta.json 路径本身。现在显式拒绝 sidecar 路径，
    要求传入数据文件路径。
#4：Sweep 对 sidecar 损坏的数据文件跳过清理。现在损坏 sidecar 会回退
    到数据文件 mtime，超期即删除。
#5：upload_file 描述仍引用已移除的 args 字段，已改为引用 script_path
    及结构化字段。
#6：Deps.UploadStore 改为值类型 uploads.Store，避免 nil 绕过上传校验；
    移除 spark_submit/upload_file 中的 nil 检查。
#7：master/queue/executor_memory 增加空字符串校验。
#8：提取 buildSparkSubmitCommand 构建 argv，消除双写参数的结构性根因。
#10：Validate 失败时记录 slog.Warn("spark_submit.unminted_path_rejected")。

新增测试：
- TestSparkSubmit_StructuredCommand：断言 argv 首行为 --master，末行
  仍为 script_path。
- TestSparkSubmit_EmptyMaster：空 master 返回错误。
- TestStore_Validate_RejectsSidecarPath：拒绝 .meta.json 路径。
- TestStore_Sweep_DeletesDataWithCorruptSidecar：损坏 sidecar 的数据文件
  被清理。

未在本提交处理：
- #9 cluster.DefaultSubmitArgs 弃用留作后续批次。

Co-Authored-By: tao.chen <93983997+taochen-ct@users.noreply.github.com>

internal/uploads/store.go

@@ -25,31 +25,31 @@ type Store struct {
 
 // New creates a Store rooted at root. It creates root with mode 0o750 if it
 // does not exist. The root must be an absolute path.
-func New(root string) (*Store, error) {
+func New(root string) (Store, error) {
 	if root == "" {
-		return nil, fmt.Errorf("uploads: root must not be empty")
+		return Store{}, fmt.Errorf("uploads: root must not be empty")
 	}
 	absRoot, err := filepath.Abs(root)
 	if err != nil {
-		return nil, fmt.Errorf("uploads: resolve root: %w", err)
+		return Store{}, fmt.Errorf("uploads: resolve root: %w", err)
 	}
 	if !filepath.IsAbs(absRoot) {
-		return nil, fmt.Errorf("uploads: root %q is not an absolute path", absRoot)
+		return Store{}, fmt.Errorf("uploads: root %q is not an absolute path", absRoot)
 	}
 
 	if err := os.MkdirAll(absRoot, 0o750); err != nil {
-		return nil, fmt.Errorf("uploads: create root: %w", err)
+		return Store{}, fmt.Errorf("uploads: create root: %w", err)
 	}
 
 	info, err := os.Stat(absRoot)
 	if err != nil {
-		return nil, fmt.Errorf("uploads: stat root: %w", err)
+		return Store{}, fmt.Errorf("uploads: stat root: %w", err)
 	}
 	if !info.IsDir() {
-		return nil, fmt.Errorf("uploads: root %q is not a directory", absRoot)
+		return Store{}, fmt.Errorf("uploads: root %q is not a directory", absRoot)
 	}
 
-	return &Store{Root: absRoot}, nil
+	return Store{Root: absRoot}, nil
 }
 
 type sidecar struct {
@@ -137,6 +137,10 @@ func (s *Store) Validate(absPath string) (string, error) {
 		return "", fmt.Errorf("uploads: path %q is not under Root", absPath)
 	}
 
+	if strings.HasSuffix(rel, ".meta.json") {
+		return "", fmt.Errorf("uploads: path %q is a sidecar, pass the data file path", absPath)
+	}
+
 	fileID := strings.TrimSuffix(rel, ".meta.json")
 	if !fileIDRegex.MatchString(fileID) {
 		return "", fmt.Errorf("uploads: path %q does not denote a valid upload", absPath)
@@ -198,13 +202,16 @@ func (s *Store) Sweep(ttl time.Duration) (deleted int, err error) {
 
 		if it.data {
 			expired := false
+			corruptSidecar := false
 			if it.meta {
 				uploadedAt, parseErr := readUploadedAt(metaPath)
 				if parseErr == nil && uploadedAt.Before(cutoff) {
 					expired = true
+				} else if parseErr != nil {
+					corruptSidecar = true
 				}
 			}
-			if !expired && !it.meta {
+			if !expired && (corruptSidecar || !it.meta) {
 				info, statErr := os.Stat(dataPath)
 				if statErr == nil && info.ModTime().Before(cutoff) {
 					expired = true

internal/uploads/store_test.go

@@ -256,3 +256,61 @@ func TestStore_Sweep_OrphanSidecarReaped(t *testing.T) {
 func timePtr(t time.Time) *time.Time {
 	return &t
 }
+
+func TestStore_Validate_RejectsSidecarPath(t *testing.T) {
+	store, err := New(t.TempDir())
+	if err != nil {
+		t.Fatalf("new store: %v", err)
+	}
+
+	fileID := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	if err := os.WriteFile(filepath.Join(store.Root, fileID), []byte("x"), 0o640); err != nil {
+		t.Fatalf("create data file: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(store.Root, fileID+".meta.json"), []byte(`{"uploaded_at":"`+time.Now().Format(time.RFC3339Nano)+`"}`), 0o600); err != nil {
+		t.Fatalf("create sidecar: %v", err)
+	}
+
+	_, err = store.Validate(filepath.Join(store.Root, fileID+".meta.json"))
+	if err == nil {
+		t.Errorf("Validate(sidecar) succeeded, want error")
+	}
+}
+
+func TestStore_Sweep_DeletesDataWithCorruptSidecar(t *testing.T) {
+	store, err := New(t.TempDir())
+	if err != nil {
+		t.Fatalf("new store: %v", err)
+	}
+
+	id := "00000000000000000000000000000005"
+	dataPath := filepath.Join(store.Root, id)
+	metaPath := dataPath + ".meta.json"
+
+	if err := os.WriteFile(dataPath, []byte("stale data"), 0o640); err != nil {
+		t.Fatalf("create data file: %v", err)
+	}
+	// Corrupt sidecar: not valid JSON.
+	if err := os.WriteFile(metaPath, []byte("not valid json"), 0o600); err != nil {
+		t.Fatalf("create sidecar: %v", err)
+	}
+	// Set mtime well in the past so the fallback triggers.
+	past := time.Now().Add(-2 * time.Hour)
+	if err := os.Chtimes(dataPath, past, past); err != nil {
+		t.Fatalf("set mtime: %v", err)
+	}
+
+	deleted, err := store.Sweep(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if deleted != 1 {
+		t.Errorf("deleted=%d, want 1", deleted)
+	}
+	if _, err := os.Stat(dataPath); !os.IsNotExist(err) {
+		t.Errorf("data file with corrupt sidecar still exists")
+	}
+	if _, err := os.Stat(metaPath); !os.IsNotExist(err) {
+		t.Errorf("corrupt sidecar still exists")
+	}
+}