ec: re-plan balance moves when a target reports no space left (#11759)

* ecbalancer: add Node.SetNoReceive to exclude a node as a move target

A node that has refused a shard copy at runtime (e.g. 'no space left' on the actual disk, issue #11753) can no longer be trusted as a destination for the remainder of a balance run, but it may still hold shards that must move away. SetNoReceive excludes the node from target selection and from receiving-side capacity accounting, while leaving it eligible as a move source.

* ec: re-plan balance moves when a target reports no space left

Moves are planned from a heartbeat snapshot, but concurrent copies let the volume server's real disk state diverge from it, so a planned target can refuse the copy with 'no space left' mid-run and abort the whole balance (issue #11753). Track the refusing targets during execution, mark them no-receive, re-collect the topology, and plan the remaining moves again. The loop terminates because a marked node can never be picked as a target again.

* volume server: return ResourceExhausted when no disk can take an EC shard

A failed EcShardsCopy carried code=Unknown even though the cause is exhausted local capacity; ResourceExhausted describes the refusal and lets clients distinguish it from internal errors. The message text is kept so message-based matching still works (the Rust volume server reports the same text under Internal).

* ec: replay applied moves onto the re-collected snapshot and surface failures

Addresses review on the re-plan loop: a fresh topology read can still describe an applied move while heartbeats lag, so journal each applied move and replay it onto the next snapshot; retry only when every failure was a 'no space left' refusal (an unrelated failure aborts instead of being hidden); and exclude refused nodes' slots from the free-space precheck so a balance with no remaining receiver fails instead of reporting success for an empty plan.

* ec: carry per-disk state when replaying applied moves

A move that landed on an explicit TargetDisk loses that attribution when replayed onto a stale snapshot, leaving the disk's free slots and shard count unchanged; the next plan could pick the same disk, hit another refusal, and over-exclude the node. Replay the shard onto the target disk's entry and move the disk-level counters, and restore the source disk's counters on removal.

* ecbalancer: add Node.SetDiskNoReceive to exclude a single disk as a target

A copy with an explicit disk_id can be refused while the node's other disks still have room; a disk_id 0 refusal already means the server searched every disk, so the scopes differ. disk.noReceive removes that disk from receiving capacity and disk selection while leaving the node and its sibling disks eligible.

* ec: record explicit-disk refusals as disk-scoped no-receive

recordMoveResult now distinguishes the two refusal scopes: TargetDisk>0 marks only (node, disk), while TargetDisk==0 marks the whole node since the server's auto-pick found no usable disk. The free-slot precheck subtracts refused disk slots, and the replan log names refused destinations as node:diskN.

* ecbalancer: skip no-receive disks in pickBestDiskOnNode

The balance phases pick the destination disk through pickBestDiskOnNode, which ranked disks by score only; a refused disk with a low shard count would be re-picked, fail again, and abort the run since no new destination was rejected.

* volume_move: export IsNoSpaceLeftError for move callers

Both the ec.balance replanner and the worker ec_balance task need to recognize the volume server's no-space refusal; keep one matcher next to the move RPCs instead of a private copy in weed/ec.

* worker: retry pinned-disk EC moves with a server-picked disk on no space

ec_balance detection always pins TargetDisk, so a full disk fails the job even when a sibling disk has room. On a no-space refusal to the pinned disk, retry the move once with TargetDisk 0 so the volume server picks a disk on the same node; the planner's node choice stands.

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/facebookgo/stack v0.0.0-20160209184415-751773369052 // indirect
 	github.com/facebookgo/stats v0.0.0-20151006221625-1b76add642e4
 	github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4 // indirect
-	github.com/fsnotify/fsnotify v1.10.1 // indirect
+	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-redsync/redsync/v4 v4.18.0
 	github.com/go-sql-driver/mysql v1.10.1
 	github.com/go-zookeeper/zk v1.0.4 // indirect
@@ -116,7 +116,7 @@ require (
 	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/Jille/raft-grpc-transport v1.6.1
 	github.com/ThreeDotsLabs/watermill v1.5.2
-	github.com/a-h/templ v0.3.1070
+	github.com/a-h/templ v0.3.1020
 	github.com/apache/cassandra-gocql-driver/v2 v2.1.2
 	github.com/apache/iceberg-go v0.7.0
 	github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3
@@ -313,7 +313,7 @@ require (
 	github.com/ProtonMail/go-srp v0.0.7 // indirect
 	github.com/PuerkitoBio/goquery v1.12.0 // indirect
 	github.com/abbot/go-http-auth v0.4.0 // indirect
-	github.com/andybalholm/brotli v1.2.6 // indirect
+	github.com/andybalholm/brotli v1.2.3 // indirect
 	github.com/andybalholm/cascadia v1.3.4 // indirect
 	github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc // indirect
 	github.com/arangodb/go-velocypack v0.0.0-20200318135517-5af53c29c67e // indirect

go.sum

@@ -118,8 +118,8 @@ github.com/ThreeDotsLabs/watermill v1.5.2 h1:0ES33Eq1jEsP/pWvtE4n8bE0bs+9Jq7boT7
 github.com/ThreeDotsLabs/watermill v1.5.2/go.mod h1:i9/968UriGphWfEbfMuYSD1qFbYRjb0mE0r+rV0FPp4=
 github.com/TomiHiltunen/geohash-golang v0.0.0-20150112065804-b3e4e625abfb h1:wumPkzt4zaxO4rHPBrjDK8iZMR41C1qs7njNqlacwQg=
 github.com/TomiHiltunen/geohash-golang v0.0.0-20150112065804-b3e4e625abfb/go.mod h1:QiYsIBRQEO+Z4Rz7GoI+dsHVneZNONvhczuA+llOZNM=
-github.com/a-h/templ v0.3.1070 h1:3iCOXHXqFXa99OQzgkUOLJbNzv/QjFGLwKZkBAg+ADA=
-github.com/a-h/templ v0.3.1070/go.mod h1:iIqB3g2RHBe/WlnPwa+Bvhh1TKXHWLRnBD9NWHOLm78=
+github.com/a-h/templ v0.3.1020 h1:ypAT/L5ySWEnZ6Zft/5yfoWXYYkhFNvEFOeeqecg4tw=
+github.com/a-h/templ v0.3.1020/go.mod h1:A2DlK61v+K+NRoGnhmYbNYVmtYHcFO5/AisMvBdDxTM=
 github.com/a1ex3/zstd-seekable-format-go/pkg v0.10.0 h1:iLDOF0rdGTrol/q8OfPIIs5kLD8XvA2q75o6Uq/tgak=
 github.com/a1ex3/zstd-seekable-format-go/pkg v0.10.0/go.mod h1:DrEWcQJjz7t5iF2duaiyhg4jyoF0kxOD6LtECNGkZ/Q=
 github.com/aalpar/deheap v1.1.2 h1:MABHLcnjqsffb8GLkUFDigqpBBxOMz0DoKM9QfELeTw=
@@ -142,8 +142,8 @@ github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktp
 github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4=
 github.com/anchore/go-lzo v0.1.1 h1:IwL/fvkdtlIrYIXck6WxZ3nb8WjjHziYYmGxlooyOnM=
 github.com/anchore/go-lzo v0.1.1/go.mod h1:3kLx0bve2oN1iDwgM1U5zGku1Tfbdb0No5qp1eL1fIk=
-github.com/andybalholm/brotli v1.2.6 h1:ftYnfj6usCp+UGV5kSJ3+chpMQgU+gJf/AxsUQ52REI=
-github.com/andybalholm/brotli v1.2.6/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
+github.com/andybalholm/brotli v1.2.3 h1:8H1qwOkl2LPfjf3YezB90JnCliZb6SInJ/OJkEbA5NQ=
+github.com/andybalholm/brotli v1.2.3/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/andybalholm/cascadia v1.3.4 h1:vM2lgh0Vru9Vwyfm4cQqWP2HHMW0u0+2PAW7Q38Qufg=
 github.com/andybalholm/cascadia v1.3.4/go.mod h1:BLRmbRjpEtNKieZOCCvYj4RqN+KRA41GBe/5O+G93kM=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
@@ -462,8 +462,8 @@ github.com/fsnotify/fsevents v0.2.0 h1:BRlvlqjvNTfogHfeBOFvSC9N0Ddy+wzQCQukyoD7o
 github.com/fsnotify/fsevents v0.2.0/go.mod h1:B3eEk39i4hz8y1zaWS/wPrAP4O6wkIl7HQwKBr1qH/w=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
-github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho=
-github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
+github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
+github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/fvbommel/sortorder v1.1.0 h1:fUmoe+HLsBTctBDoaBwpQo5N+nrCp8g/BjKb/6ZQmYw=
 github.com/fvbommel/sortorder v1.1.0/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
 github.com/gabriel-vasile/mimetype v1.4.13 h1:46nXokslUBsAJE/wMsp5gtO500a4F3Nkz9Ufpk2AcUM=

weed/command/mini.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -64,8 +65,9 @@ var (
 	miniS3Options     S3Options
 	miniWebDavOptions WebDavOption
 	miniAdminOptions  AdminOptions
-	// Track which port flags were explicitly passed on CLI before config file is applied
-	explicitPortFlags             map[string]bool
+	// Flags explicitly provided on the command line or via WEED_* env vars,
+	// captured before the saved options file is applied
+	explicitFlags                 map[string]bool
 	miniEnableWebDAV              *bool
 	miniEnableS3                  *bool
 	miniEnableAdminUI             *bool
@@ -376,6 +378,19 @@ The S3 endpoint is available at http://localhost:8333. For client
 configuration and IAM setup, see the project documentation or use the
 Admin UI (http://localhost:23646) to manage users and policies.
 
+Notes:
+- Values mini computes itself (currently master.volumeSizeLimitMB) are
+  pinned in <dir>/mini.options and applied again on the next start to
+  keep them stable; applied values are printed at startup. Command-line
+  flags are never saved and always win over saved values. Edit or
+  delete mini.options to reset a computed value.
+- The official Docker image ships /etc/seaweedfs/filer.toml, which keeps
+  the filer metadata in /data/filerldb2 regardless of -dir; mini warns
+  at startup when a configured store dir is outside -dir. If you set
+  -dir to another path, keep /data on a persistent volume as well.
+  Otherwise the metadata is lost when the container is recreated, and
+  buckets list no objects although the volume files are still in -dir.
+
 `,
 }
 
@@ -823,8 +838,8 @@ func findAvailablePortOnIP(ip string, startPort int, maxAttempts int, reservedPo
 // If the port is not available, it finds the next available port and updates the pointer
 // The reservedPorts map contains ports that should not be allocated (for gRPC collision avoidance)
 func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reservedPorts map[int]bool, flagName string) error {
-	// Check if this port was explicitly specified by the user (from CLI, before config file was applied)
-	isExplicitPort := explicitPortFlags[flagName]
+	// Check if this port was explicitly specified by the user (from CLI or env, before the config file was applied)
+	isExplicitPort := explicitFlags[flagName]
 
 	if *portPtr == 0 {
 		return nil
@@ -1072,10 +1087,16 @@ func initializeGrpcPortsOnIP(bindIp string) {
 	}
 }
 
+// miniOptionsFilePath returns the path of the mini.options file under the
+// first -dir folder.
+func miniOptionsFilePath(dataFolder string) string {
+	return filepath.Join(util.ResolvePath(util.StringSplit(dataFolder, ",")[0]), "mini.options")
+}
+
 // loadMiniConfigurationFile reads the mini.options file and returns parsed options
 // File format: one option per line, without leading dash (e.g., "ip=127.0.0.1")
 func loadMiniConfigurationFile(dataFolder string) (map[string]string, error) {
-	configFile := filepath.Join(util.ResolvePath(util.StringSplit(dataFolder, ",")[0]), "mini.options")
+	configFile := miniOptionsFilePath(dataFolder)
 
 	options := make(map[string]string)
 
@@ -1123,31 +1144,112 @@ func loadMiniConfigurationFile(dataFolder string) (map[string]string, error) {
 	return options, nil
 }
 
-// applyConfigFileOptions sets command-line flags from loaded configuration file
-func applyConfigFileOptions(options map[string]string) {
-	for key, value := range options {
-		// Skip port flags that were explicitly passed on CLI
-		if explicitPortFlags[key] {
-			glog.V(2).Infof("Skipping config file option %s=%s (explicitly specified on command line)", key, value)
+// applyConfigFileOptions sets command-line flags from loaded configuration
+// file. Options for flags the user provided explicitly (command line or
+// WEED_* env var) are skipped, so passing a flag always wins over the saved
+// value — including passing it at its default value. Returns a map of the
+// applied keys to their display values (secrets redacted).
+func applyConfigFileOptions(options map[string]string) map[string]string {
+	keys := make([]string, 0, len(options))
+	for key := range options {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+
+	applied := make(map[string]string)
+	for _, key := range keys {
+		value := options[key]
+		// Redact secrets (admin.password, admin.readOnlyPassword, ...) in
+		// logs and the startup message; only flag.Value.Set sees the value.
+		display := value
+		if isSensitiveOption(key) {
+			display = "*****"
+		}
+		if explicitFlags[key] {
+			glog.V(2).Infof("Skipping config file option %s=%s (explicitly specified)", key, display)
 			continue
 		}
-		// Set the flag value if it hasn't been explicitly set on command line
 		flag := cmdMini.Flag.Lookup(key)
-		if flag != nil {
-			// Only set if not already set (by command line)
-			if flag.Value.String() == flag.DefValue {
-				if err := flag.Value.Set(value); err != nil {
-					glog.Warningf("Failed to apply config file option: %s=%s: %v", key, value, err)
-				} else {
-					glog.V(2).Infof("Applied config file option: %s=%s", key, value)
+		if flag == nil {
+			continue
+		}
+		if err := flag.Value.Set(value); err != nil {
+			glog.Warningf("Failed to apply config file option: %s=%s: %v", key, display, err)
+			continue
+		}
+		glog.V(2).Infof("Applied config file option: %s=%s", key, display)
+		applied[key] = display
+	}
+	return applied
+}
+
+// isSensitiveOption reports whether a saved option holds a secret that must
+// not appear in logs or the startup message.
+func isSensitiveOption(key string) bool {
+	k := strings.ToLower(key)
+	return strings.Contains(k, "password") || strings.Contains(k, "secret") || strings.Contains(k, "token")
+}
+
+// warnIfFilerStoreOutsideDir warns when a filer.toml configures a local
+// filer store outside the -dir folders and the master meta folder. When a
+// filer.toml is found the filer skips defaultLevelDbDirectory, so metadata
+// can silently land on a path the volume data doesn't use — in a container
+// that path is lost on recreate unless it is persisted separately.
+func warnIfFilerStoreOutsideDir(dataFolders string, metaFolder string) {
+	if !util.LoadConfiguration("filer", false) {
+		return
+	}
+	v := util.GetViper()
+	folders := append(util.StringSplit(dataFolders, ","), metaFolder)
+	inside := func(storeDir string) bool {
+		// The filer stores open the configured dir verbatim, so absolutize it
+		// the same way (relative to the working directory) rather than
+		// expanding "~", which the store never expands.
+		resolved, err := filepath.Abs(storeDir)
+		if err != nil {
+			return false
+		}
+		for _, folder := range folders {
+			base, err := filepath.Abs(util.ResolvePath(folder))
+			if err != nil {
+				continue
+			}
+			// Resolve symlinks only when both sides exist, so a not-yet-
+			// created store dir under a symlinked -dir still compares
+			// lexically against the same spelling.
+			r, b := resolved, base
+			if er, err := filepath.EvalSymlinks(resolved); err == nil {
+				if eb, err := filepath.EvalSymlinks(base); err == nil {
+					r, b = er, eb
 				}
 			}
+			if rel, err := filepath.Rel(b, r); err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
+				return true
+			}
+		}
+		return false
+	}
+	// Local filer stores keep metadata under a "dir" key; check both the
+	// default store (leveldb2.enabled) and path-specific stores
+	// (leveldb2.<id>.enabled, keyed under <section>.location).
+	localStores := map[string]bool{"leveldb": true, "leveldb2": true, "leveldb3": true}
+	for _, key := range v.AllKeys() {
+		if !strings.HasSuffix(key, ".enabled") || !v.GetBool(key) {
+			continue
+		}
+		section := strings.TrimSuffix(key, ".enabled")
+		if !localStores[strings.Split(section, ".")[0]] {
+			continue
+		}
+		if storeDir := v.GetString(section + ".dir"); storeDir != "" && !inside(storeDir) {
+			glog.Warningf("filer store %s uses dir %s from filer.toml, outside -dir %s; persist that path or filer metadata will be lost when the container is recreated", section, storeDir, dataFolders)
 		}
 	}
 }
 
-// saveMiniConfiguration saves the current mini configuration to a file
-// The file format uses option=value format without leading dashes
+// saveMiniConfiguration writes values mini computed itself to mini.options.
+// Command-line flags are never saved, so a dropped flag cannot resurrect on
+// the next start; the file exists to pin environment-derived values.
 func saveMiniConfiguration(dataFolder string) error {
 	configDir := util.ResolvePath(util.StringSplit(dataFolder, ",")[0])
 	if err := os.MkdirAll(configDir, 0755); err != nil {
@@ -1159,29 +1261,15 @@ func saveMiniConfiguration(dataFolder string) error {
 
 	var sb strings.Builder
 	sb.WriteString("#!/bin/bash\n")
-	sb.WriteString("# Mini server configuration\n")
+	sb.WriteString("# Mini server auto-computed options\n")
 	sb.WriteString("# Format: option=value (no leading dash)\n")
-	sb.WriteString("# This file is loaded on startup if it exists\n\n")
+	sb.WriteString("# This file is loaded on startup if it exists\n")
+	sb.WriteString("# Command-line flags are not saved; only values mini computed itself.\n\n")
 
-	// Collect all flags that were explicitly passed (except "dir")
-	cmdMini.Flag.Visit(func(f *flag.Flag) {
-		// Skip the "dir" option - it's environment-specific
-		if f.Name == "dir" {
-			return
-		}
-		value := f.Value.String()
-		// Quote the value if it contains spaces
-		if strings.Contains(value, " ") {
-			sb.WriteString(fmt.Sprintf("%s=\"%s\"\n", f.Name, value))
-		} else {
-			sb.WriteString(fmt.Sprintf("%s=%s\n", f.Name, value))
-		}
-	})
-
-	// Add auto-calculated volume size if it was computed
+	// Pin the auto-calculated volume size so it stays stable across restarts
+	// even as free disk space changes. Delete this line to force
+	// recalculation on the next startup.
 	if !isFlagPassed("master.volumeSizeLimitMB") && miniMasterOptions.volumeSizeLimitMB != nil {
-		sb.WriteString(fmt.Sprintf("\n# Auto-calculated volume size based on total disk capacity\n"))
-		sb.WriteString(fmt.Sprintf("# Delete this line to force recalculation on next startup\n"))
 		sb.WriteString(fmt.Sprintf("master.volumeSizeLimitMB=%d\n", *miniMasterOptions.volumeSizeLimitMB))
 	}
 
@@ -1203,21 +1291,32 @@ func runMini(cmd *Command, args []string) bool {
 	// can opt back into verbose output.
 	quietMiniLogs()
 
-	// Capture which port flags were explicitly passed on CLI BEFORE config file is applied
-	// This is necessary to distinguish user-specified ports from defaults or config file options
-	explicitPortFlags = make(map[string]bool)
-	portFlagNames := []string{"master.port", "filer.port", "volume.port", "s3.port", "s3.port.iceberg", "webdav.port", "admin.port", "s3.iam.readOnly"}
-	for _, flagName := range portFlagNames {
-		explicitPortFlags[flagName] = isFlagPassed(flagName)
-	}
+	// Capture which flags were explicitly provided on the command line or via
+	// WEED_* env vars BEFORE the saved options file is applied. Saved options
+	// only fill in flags the user did not provide this run, so passing a flag
+	// at its default value also resets a saved value.
+	explicitFlags = make(map[string]bool)
+	cmdMini.Flag.Visit(func(f *flag.Flag) {
+		explicitFlags[f.Name] = true
+	})
 
-	// Load configuration from file if it exists
+	// Load configuration from file if it exists. Capture the path before
+	// applying: a saved "dir" option can change miniDataFolders.
+	optionsFile := miniOptionsFilePath(*miniDataFolders)
 	configOptions, err := loadMiniConfigurationFile(*miniDataFolders)
 	if err != nil {
 		glog.Warningf("Error loading configuration file: %v", err)
 	}
-	// Apply loaded options to flags (CLI flags will override these)
-	applyConfigFileOptions(configOptions)
+	// Apply loaded options to flags (explicitly provided flags win)
+	appliedOptions := applyConfigFileOptions(configOptions)
+	if len(appliedOptions) > 0 {
+		pairs := make([]string, 0, len(appliedOptions))
+		for k, v := range appliedOptions {
+			pairs = append(pairs, k+"="+v)
+		}
+		sort.Strings(pairs)
+		fmt.Printf("  Applying %d saved option(s) from %s: %s\n", len(appliedOptions), optionsFile, strings.Join(pairs, " "))
+	}
 
 	if *miniOptions.debug {
 		grace.StartDebugServer(*miniOptions.debugPort)
@@ -1346,10 +1445,14 @@ func runMini(cmd *Command, args []string) bool {
 		glog.Fatalf("Check Meta Folder (-dir=\"%s\") Writable: %s", *miniMasterOptions.metaFolder, err)
 	}
 	miniFilerOptions.defaultLevelDbDirectory = miniMasterOptions.metaFolder
+	warnIfFilerStoreOutsideDir(*miniDataFolders, *miniMasterOptions.metaFolder)
 
-	// Calculate and set optimal volume size limit based on available disk space
-	// Only auto-calculate if user didn't explicitly specify a value via -master.volumeSizeLimitMB
-	if !isFlagPassed("master.volumeSizeLimitMB") {
+	// Calculate and set optimal volume size limit based on available disk space.
+	// Skip when the user passed -master.volumeSizeLimitMB explicitly, and when
+	// the value was loaded from mini.options — file-applied flags are not in
+	// Visit, and the saved pin must survive restarts.
+	_, sizeLimitFromFile := appliedOptions["master.volumeSizeLimitMB"]
+	if !isFlagPassed("master.volumeSizeLimitMB") && !sizeLimitFromFile {
 		// User didn't override, use auto-calculated value
 		// The -dir flag can accept comma-separated directories; use the first one for disk space calculation.
 		// miniDataFolders was already tilde-resolved at the top of runMini.

weed/ec/ec_balance_no_space_test.go

@@ -0,0 +1,359 @@
+package ec
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// ecBalanceFakeServer answers the EC-shard RPC subset a balance run drives:
+// copy (rejected when full), mount, inventory, unmount, delete. The disk state
+// a real server tracks lives in shards; what the planner saw is rebuilt from it
+// through FetchTopology, so a landed move is visible to the next plan.
+type ecBalanceFakeServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+
+	mu        sync.Mutex
+	shards    map[uint32]map[uint32]bool // vid -> shardId -> present
+	full      bool                       // VolumeEcShardsCopy answers "no space left"
+	copyCalls int
+	grpcAddr  string
+	stop      func()
+}
+
+func (f *ecBalanceFakeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.copyCalls++
+	if f.full {
+		return nil, fmt.Errorf("no space left")
+	}
+	return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil
+}
+
+func (f *ecBalanceFakeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for _, sid := range req.ShardIds {
+		f.shards[req.VolumeId][sid] = true
+	}
+	return &volume_server_pb.VolumeEcShardsMountResponse{}, nil
+}
+
+func (f *ecBalanceFakeServer) VolumeEcShardsInfo(ctx context.Context, req *volume_server_pb.VolumeEcShardsInfoRequest) (*volume_server_pb.VolumeEcShardsInfoResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	resp := &volume_server_pb.VolumeEcShardsInfoResponse{}
+	for sid := range f.shards[req.VolumeId] {
+		resp.EcShardInfos = append(resp.EcShardInfos, &volume_server_pb.EcShardInfo{
+			VolumeId: req.VolumeId,
+			ShardId:  sid,
+			Size:     1,
+		})
+	}
+	return resp, nil
+}
+
+func (f *ecBalanceFakeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) {
+	return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil
+}
+
+func (f *ecBalanceFakeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for _, sid := range req.ShardIds {
+		delete(f.shards[req.VolumeId], sid)
+	}
+	return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil
+}
+
+// nodeId returns the "host:httpPort.grpcPort" node id form the cluster uses, so
+// ToGrpcAddress resolves to this server's listener.
+func (f *ecBalanceFakeServer) nodeId() string {
+	host, port, err := net.SplitHostPort(f.grpcAddr)
+	if err != nil {
+		return f.grpcAddr
+	}
+	return fmt.Sprintf("%s:1.%s", host, port)
+}
+
+func (f *ecBalanceFakeServer) shardCount(vid uint32) int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.shards[vid])
+}
+
+func startEcBalanceFakeServer(t *testing.T, vid uint32, present ...uint32) *ecBalanceFakeServer {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	f := &ecBalanceFakeServer{
+		shards:   map[uint32]map[uint32]bool{vid: {}},
+		grpcAddr: listener.Addr().String(),
+	}
+	for _, sid := range present {
+		f.shards[vid][sid] = true
+	}
+
+	server := grpc.NewServer()
+	volume_server_pb.RegisterVolumeServerServer(server, f)
+	go func() { _ = server.Serve(listener) }()
+	f.stop = server.Stop
+	t.Cleanup(server.Stop)
+	return f
+}
+
+// fetchEcBalanceTopology builds a FetchTopology that reports the fake servers'
+// live shard inventory.
+func fetchEcBalanceTopology(servers []*ecBalanceFakeServer) func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
+	return func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
+		var nodes []*master_pb.DataNodeInfo
+		for _, f := range servers {
+			di := &master_pb.DiskInfo{Type: string(types.HardDriveType), MaxVolumeCount: 100}
+			f.mu.Lock()
+			for v, set := range f.shards {
+				if len(set) == 0 {
+					continue
+				}
+				si := erasure_coding.NewShardsInfo()
+				for sid := range set {
+					si.Set(erasure_coding.NewShardInfo(erasure_coding.ShardId(sid), 1))
+				}
+				di.EcShardInfos = append(di.EcShardInfos, &master_pb.VolumeEcShardInformationMessage{
+					Id:          v,
+					Collection:  "c1",
+					EcIndexBits: si.Bitmap(),
+					ShardSizes:  si.SizesInt64(),
+					DiskType:    string(types.HardDriveType),
+				})
+			}
+			f.mu.Unlock()
+			nodes = append(nodes, &master_pb.DataNodeInfo{
+				Id:        f.nodeId(),
+				DiskInfos: map[string]*master_pb.DiskInfo{string(types.HardDriveType): di},
+			})
+		}
+		return &master_pb.TopologyInfo{
+			DataCenterInfos: []*master_pb.DataCenterInfo{{
+				Id: "dc1",
+				RackInfos: []*master_pb.RackInfo{{
+					Id:            "r1",
+					DataNodeInfos: nodes,
+				}},
+			}},
+		}, 0, nil
+	}
+}
+
+// TestEcBalanceReplanAfterNoSpaceLeft reproduces seaweedfs/seaweedfs#11753:
+// the planner works from heartbeat free-slot counts that can disagree with the
+// volume server's own check, so a planned target can reject the copy with
+// "no space left" mid-run. The balance must re-plan without that node as a
+// target rather than fail the whole run.
+func TestEcBalanceReplanAfterNoSpaceLeft(t *testing.T) {
+	const vid = uint32(1)
+	var allShards []uint32
+	for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ {
+		allShards = append(allShards, uint32(sid))
+	}
+
+	full := startEcBalanceFakeServer(t, vid)
+	accept := startEcBalanceFakeServer(t, vid)
+	source := startEcBalanceFakeServer(t, vid, allShards...)
+	full.full = true
+
+	env := &Env{
+		GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		FetchTopology:  fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full, accept}),
+	}
+
+	err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil)
+	if err != nil {
+		t.Fatalf("balance should re-plan around the full target, got: %v", err)
+	}
+	if full.copyCalls == 0 {
+		t.Fatalf("expected the full server to have rejected at least one copy")
+	}
+	if got := full.shardCount(vid); got != 0 {
+		t.Fatalf("full server should receive no shards, holds %d", got)
+	}
+	total := source.shardCount(vid) + full.shardCount(vid) + accept.shardCount(vid)
+	if total != len(allShards) {
+		t.Fatalf("shards lost or duplicated: total %d, want %d", total, len(allShards))
+	}
+	if source.shardCount(vid) == len(allShards) {
+		t.Fatalf("no shards were moved off the source")
+	}
+}
+
+// TestEcBalanceReplanAfterNoSpaceLeftStaleSnapshot serves the pre-move
+// topology on every collection after the first, simulating a master heartbeat
+// that lags the landed moves. The re-plan must replay its completed moves onto
+// the stale snapshot instead of re-moving shards off the deleted source.
+func TestEcBalanceReplanAfterNoSpaceLeftStaleSnapshot(t *testing.T) {
+	const vid = uint32(1)
+	var allShards []uint32
+	for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ {
+		allShards = append(allShards, uint32(sid))
+	}
+
+	full := startEcBalanceFakeServer(t, vid)
+	accept := startEcBalanceFakeServer(t, vid)
+	source := startEcBalanceFakeServer(t, vid, allShards...)
+	full.full = true
+
+	live := fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full, accept})
+	var stale *master_pb.TopologyInfo
+	calls := 0
+	fetchTopology := func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
+		topo, size, err := live(delay)
+		if err != nil {
+			return nil, 0, err
+		}
+		calls++
+		if stale == nil {
+			stale = topo
+		}
+		if calls > 1 {
+			return stale, size, nil
+		}
+		return topo, size, nil
+	}
+
+	env := &Env{
+		GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		FetchTopology:  fetchTopology,
+	}
+
+	err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil)
+	if err != nil {
+		t.Fatalf("balance should re-plan around the full target, got: %v", err)
+	}
+	if calls < 2 {
+		t.Fatalf("expected a re-plan, topology collected %d time(s)", calls)
+	}
+	// Each shard lands on the accepting server exactly once; a stale re-plan
+	// would copy already-moved shards again.
+	if got, want := accept.copyCalls, accept.shardCount(vid); got != want {
+		t.Fatalf("accepting server saw %d copies for %d held shards: completed moves were re-planned", got, want)
+	}
+	total := source.shardCount(vid) + full.shardCount(vid) + accept.shardCount(vid)
+	if total != len(allShards) {
+		t.Fatalf("shards lost or duplicated: total %d, want %d", total, len(allShards))
+	}
+}
+
+// TestEcBalanceNoReceivingTargetFails surfaces the refusal when every
+// destination that could take a shard has rejected a copy, rather than
+// reporting success for a plan that moved nothing.
+func TestEcBalanceNoReceivingTargetFails(t *testing.T) {
+	const vid = uint32(1)
+	var allShards []uint32
+	for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ {
+		allShards = append(allShards, uint32(sid))
+	}
+
+	full := startEcBalanceFakeServer(t, vid)
+	source := startEcBalanceFakeServer(t, vid, allShards...)
+	full.full = true
+
+	env := &Env{
+		GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		FetchTopology:  fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full}),
+	}
+
+	err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil)
+	if err == nil {
+		t.Fatalf("balance with no receiving target should fail")
+	}
+	if !strings.Contains(err.Error(), "no space left") {
+		t.Fatalf("expected the refusal to surface, got: %v", err)
+	}
+}
+
+// TestApplyCompletedMovesReplaysDiskState checks that replaying a landed move
+// onto a stale snapshot carries the shard's disk attribution and the per-disk
+// slot accounting, so a re-plan does not treat the just-filled disk as empty.
+func TestApplyCompletedMovesReplaysDiskState(t *testing.T) {
+	const vid = uint32(1)
+
+	source := newEcNode("dc1", "r1", "dn1", 99)
+	source.Info.DiskInfos["hdd"] = &master_pb.DiskInfo{
+		Type: "hdd",
+		EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{
+			Id:          vid,
+			Collection:  "c1",
+			DiskId:      1,
+			DiskType:    "hdd",
+			EcIndexBits: uint32(erasure_coding.ShardBits(0).Set(5)),
+		}},
+	}
+	si := erasure_coding.NewShardsInfo()
+	si.Set(erasure_coding.NewShardInfo(5, 1))
+	source.Disks = map[uint32]*EcDisk{
+		1: {DiskId: 1, DiskType: "hdd", FreeEcSlots: 99, EcShardCount: 1, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{1: si}},
+	}
+
+	target := newEcNode("dc1", "r1", "dn2", 100)
+	target.Disks = map[uint32]*EcDisk{
+		1: {DiskId: 1, DiskType: "hdd", FreeEcSlots: 10, EcShardCount: 0, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{}},
+		2: {DiskId: 2, DiskType: "hdd", FreeEcSlots: 10, EcShardCount: 0, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{}},
+	}
+
+	nodes := []*EcNode{source, target}
+	move := ecbalancer.Move{
+		VolumeID:   vid,
+		ShardID:    5,
+		Collection: "c1",
+		SourceNode: "dn1",
+		TargetNode: "dn2",
+		TargetDisk: 2,
+		Phase:      "within_rack",
+	}
+	applyCompletedMoves(nodes, []ecbalancer.Move{move}, types.HardDriveType)
+
+	if ecNodeHasShard(source, needle.VolumeId(vid), 5) {
+		t.Fatalf("source still holds shard 5")
+	}
+	if d := source.Disks[1]; d.EcShardCount != 0 || d.FreeEcSlots != 100 {
+		t.Fatalf("source disk counts not restored: %+v", d)
+	}
+	if !ecNodeHasShard(target, needle.VolumeId(vid), 5) {
+		t.Fatalf("target missing shard 5")
+	}
+	if d := target.Disks[2]; d.EcShardCount != 1 || d.FreeEcSlots != 9 {
+		t.Fatalf("target disk 2 not charged: %+v", d)
+	}
+	if d := target.Disks[1]; d.EcShardCount != 0 {
+		t.Fatalf("shard attributed to wrong disk: %+v", d)
+	}
+	if target.FreeEcSlot != 99 {
+		t.Fatalf("target free slots = %d, want 99", target.FreeEcSlot)
+	}
+
+	// Replaying again (or onto a fresher snapshot) must not double-count.
+	applyCompletedMoves(nodes, []ecbalancer.Move{move}, types.HardDriveType)
+	if d := target.Disks[2]; d.EcShardCount != 1 || d.FreeEcSlots != 9 {
+		t.Fatalf("double-applied move: %+v", d)
+	}
+	if target.FreeEcSlot != 99 {
+		t.Fatalf("target free slots = %d, want 99", target.FreeEcSlot)
+	}
+}

weed/ec/ec_common.go

@@ -8,6 +8,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -762,6 +763,26 @@ type ecBalancer struct {
 	// migratingVolumeIds are mid-encode volumes whose shards are ingested from
 	// every disk-type bucket; see EcBalance.
 	migratingVolumeIds map[uint32]bool
+	// noReceiveTargets are node ids the planner may not pick as destinations;
+	// noReceiveDisks narrows that to individual disks on a node: a server-side
+	// auto-pick refusal means no disk qualifies, but an explicit-disk refusal
+	// only convicts that disk. See EcBalance.
+	noReceiveTargets map[string]bool
+	noReceiveDisks   map[string]map[uint32]bool
+	// mu guards the execution journal written by concurrent move tasks.
+	mu sync.Mutex
+	// rejectedTargets and rejectedDisks collect the destinations that refused a
+	// copy this attempt; completedMoves records applied moves so a retry can
+	// replay them onto a stale-heartbeat snapshot; sawOtherErr marks any
+	// failure that is not a "no space left" refusal and makes the run
+	// non-retryable.
+	rejectedTargets map[string]bool
+	rejectedDisks   map[string]map[uint32]bool
+	completedMoves  []ecbalancer.Move
+	sawOtherErr     bool
+	// plannedMoves is the move count of the last plan, for the empty-replan
+	// check in EcBalance.
+	plannedMoves int
 }
 
 // excludeNodes is a set of server addresses kept out of the balance as copy/move
@@ -782,32 +803,6 @@ type ecBalancer struct {
 // so a plain ec.balance -diskType=X never drags deliberately tiered shards of
 // other types onto X disks.
 func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, ioBytePerSecond int64, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId, migratingVolumeIds []needle.VolumeId) (err error) {
-	// collect all ec nodes
-	allEcNodes, totalFreeEcSlots, err := CollectEcNodesForDC(env, dc, diskType)
-	if err != nil {
-		return err
-	}
-
-	// Drop excluded nodes (and the slots they contribute) before planning so they
-	// can be neither a target nor a source for any move this balance plans.
-	if len(excludeNodes) > 0 {
-		kept := allEcNodes[:0]
-		var excludedFreeSlots int
-		for _, en := range allEcNodes {
-			if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.Info)]; skip {
-				excludedFreeSlots += en.FreeEcSlot
-				glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.Info.Id)
-				continue
-			}
-			kept = append(kept, en)
-		}
-		allEcNodes = kept
-		totalFreeEcSlots -= excludedFreeSlots
-	}
-
-	if totalFreeEcSlots < 1 {
-		return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
-	}
 
 	var volumeIdFilter map[uint32]bool
 	if len(volumeIds) > 0 {
@@ -824,22 +819,277 @@ func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *su
 		}
 	}
 
-	ecb := &ecBalancer{
-		env:                env,
-		ecNodes:            allEcNodes,
-		replicaPlacement:   ecReplicaPlacement,
-		applyBalancing:     applyBalancing,
-		maxParallelization: maxParallelization,
-		ioBytePerSecond:    ioBytePerSecond,
-		diskType:           diskType,
-		volumeIds:          volumeIdFilter,
-		migratingVolumeIds: migrating,
-	}
+	// The planner's free-slot numbers come from heartbeat snapshots that can lag
+	// the volume server's own accounting, so a planned target can refuse a copy
+	// with "no space left" while shards move concurrently (issue #11753). The
+	// server's refusal is authoritative: re-collect the topology, mark the
+	// refusing targets no-receive, and plan again. They stay usable as sources.
+	noReceiveTargets := make(map[string]bool)
+	noReceiveDisks := make(map[string]map[uint32]bool)
+	var completed []ecbalancer.Move
+	var lastErr error
+	for {
+		// collect all ec nodes
+		allEcNodes, totalFreeEcSlots, err := CollectEcNodesForDC(env, dc, diskType)
+		if err != nil {
+			return err
+		}
 
-	if len(collections) == 0 {
-		glog.V(1).Infof("WARNING: No collections to balance EC volumes across.\n")
+		// Drop excluded nodes (and the slots they contribute) before planning so they
+		// can be neither a target nor a source for any move this balance plans.
+		if len(excludeNodes) > 0 {
+			kept := allEcNodes[:0]
+			var excludedFreeSlots int
+			for _, en := range allEcNodes {
+				if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.Info)]; skip {
+					excludedFreeSlots += en.FreeEcSlot
+					glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.Info.Id)
+					continue
+				}
+				kept = append(kept, en)
+			}
+			allEcNodes = kept
+			totalFreeEcSlots -= excludedFreeSlots
+		}
+		// A refused destination's slots cannot take a copy, so they do not
+		// count toward the free-space precheck.
+		for _, en := range allEcNodes {
+			if noReceiveTargets[en.Info.Id] {
+				totalFreeEcSlots -= en.FreeEcSlot
+				continue
+			}
+			for diskId := range noReceiveDisks[en.Info.Id] {
+				if d := en.Disks[diskId]; d != nil {
+					totalFreeEcSlots -= d.FreeEcSlots
+				}
+			}
+		}
+
+		if totalFreeEcSlots < 1 {
+			if lastErr != nil {
+				return fmt.Errorf("no receiving targets remain: %w", lastErr)
+			}
+			return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
+		}
+
+		// A fresh collection can still describe an applied move through a
+		// heartbeat that predates its source delete; replay the journal.
+		applyCompletedMoves(allEcNodes, completed, diskType)
+
+		ecb := &ecBalancer{
+			env:                env,
+			ecNodes:            allEcNodes,
+			replicaPlacement:   ecReplicaPlacement,
+			applyBalancing:     applyBalancing,
+			maxParallelization: maxParallelization,
+			ioBytePerSecond:    ioBytePerSecond,
+			diskType:           diskType,
+			volumeIds:          volumeIdFilter,
+			migratingVolumeIds: migrating,
+			noReceiveTargets:   noReceiveTargets,
+			noReceiveDisks:     noReceiveDisks,
+		}
+
+		if len(collections) == 0 {
+			glog.V(1).Infof("WARNING: No collections to balance EC volumes across.\n")
+		}
+		err = ecb.balance(collections)
+		if !applyBalancing {
+			return err
+		}
+		if err == nil {
+			if lastErr != nil && ecb.plannedMoves == 0 {
+				return fmt.Errorf("no receiving targets remain: %w", lastErr)
+			}
+			return nil
+		}
+
+		// Replan while the failed attempt surfaced newly-refusing targets; a node
+		// once marked can no longer be picked, so the loop terminates. Any other
+		// failure aborts instead of being hidden by the retry.
+		newRejects := ecb.mergeRejectedTargets(noReceiveTargets, noReceiveDisks)
+		if len(newRejects) == 0 || ecb.hasOtherError() {
+			return err
+		}
+		lastErr = err
+		completed = append(completed, ecb.completedMoves...)
+		fmt.Printf("ec.balance: %v reported no space left; re-collecting topology and re-planning without them as targets\n", newRejects)
+	}
+}
+
+// applyCompletedMoves replays already-applied moves onto a freshly collected
+// node list: heartbeats can lag, so the snapshot may still show a moved shard
+// on its deleted source or lack it on its mounted target.
+func applyCompletedMoves(ecNodes []*EcNode, moves []ecbalancer.Move, diskType types.DiskType) {
+	if len(moves) == 0 {
+		return
+	}
+	byID := make(map[string]*EcNode, len(ecNodes))
+	for _, en := range ecNodes {
+		byID[en.Info.Id] = en
+	}
+	for _, m := range moves {
+		vid := needle.VolumeId(m.VolumeID)
+		shardId := erasure_coding.ShardId(m.ShardID)
+		if src := byID[m.SourceNode]; src != nil {
+			removeEcShard(src, vid, shardId)
+		}
+		if m.Phase == "dedup" {
+			continue
+		}
+		if dst := byID[m.TargetNode]; dst != nil && !ecNodeHasShard(dst, vid, shardId) {
+			addEcShard(dst, m.Collection, vid, shardId, m.TargetDisk, diskType)
+		}
+	}
+}
+
+func ecNodeHasShard(ecNode *EcNode, vid needle.VolumeId, shardId erasure_coding.ShardId) bool {
+	for _, diskInfo := range ecNode.Info.DiskInfos {
+		for _, eci := range diskInfo.EcShardInfos {
+			if needle.VolumeId(eci.Id) == vid && erasure_coding.ShardBits(eci.EcIndexBits).Has(shardId) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// removeEcShard drops the shard from the node model, including the per-disk
+// bookkeeping of the disk that reported it.
+func removeEcShard(en *EcNode, vid needle.VolumeId, shardId erasure_coding.ShardId) {
+	for _, diskInfo := range en.Info.DiskInfos {
+		for _, eci := range diskInfo.EcShardInfos {
+			if needle.VolumeId(eci.Id) != vid || !erasure_coding.ShardBits(eci.EcIndexBits).Has(shardId) {
+				continue
+			}
+			if d := en.Disks[eci.DiskId]; d != nil {
+				if si := d.EcShards[vid]; si != nil {
+					si.Delete(shardId)
+				}
+				d.EcShardCount--
+				d.FreeEcSlots++
+			}
+		}
+	}
+	en.DeleteEcVolumeShards(vid, []erasure_coding.ShardId{shardId})
+}
+
+// addEcShard places the shard on the disk the move used when that disk is
+// known; otherwise (TargetDisk 0, server-picked) it falls back to the node's
+// disk-type bucket.
+func addEcShard(en *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, diskId uint32, diskType types.DiskType) {
+	d := en.Disks[diskId]
+	if diskId == 0 || d == nil {
+		en.AddEcVolumeShards(vid, collection, []erasure_coding.ShardId{shardId}, diskType)
+		return
+	}
+	diskInfo := en.Info.DiskInfos[string(diskType)]
+	if diskInfo == nil {
+		diskInfo = &master_pb.DiskInfo{Type: string(diskType)}
+		en.Info.DiskInfos[string(diskType)] = diskInfo
+	}
+	var eci *master_pb.VolumeEcShardInformationMessage
+	for _, e := range diskInfo.EcShardInfos {
+		if needle.VolumeId(e.Id) == vid && e.DiskId == diskId {
+			eci = e
+			break
+		}
+	}
+	if eci == nil {
+		eci = &master_pb.VolumeEcShardInformationMessage{
+			Id:         uint32(vid),
+			Collection: collection,
+			DiskId:     diskId,
+			DiskType:   string(diskType),
+		}
+		diskInfo.EcShardInfos = append(diskInfo.EcShardInfos, eci)
+	}
+	si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci)
+	before := si.Count()
+	si.Set(erasure_coding.NewShardInfo(shardId, 0))
+	eci.EcIndexBits = si.Bitmap()
+	eci.ShardSizes = si.SizesInt64()
+	added := si.Count() - before
+
+	en.FreeEcSlot -= added
+	d.EcShardCount += added
+	d.FreeEcSlots -= added
+	if d.EcShards == nil {
+		d.EcShards = make(map[needle.VolumeId]*erasure_coding.ShardsInfo)
+	}
+	if d.EcShards[vid] == nil {
+		d.EcShards[vid] = erasure_coding.NewShardsInfo()
+	}
+	d.EcShards[vid].Set(erasure_coding.NewShardInfo(shardId, 0))
+}
+
+// mergeRejectedTargets folds the destinations this run saw refuse into the
+// no-receive sets and returns the newly added ids; an empty result means no
+// progress is possible.
+func (ecb *ecBalancer) mergeRejectedTargets(noReceiveNodes map[string]bool, noReceiveDisks map[string]map[uint32]bool) (added []string) {
+	ecb.mu.Lock()
+	defer ecb.mu.Unlock()
+	for id := range ecb.rejectedTargets {
+		if !noReceiveNodes[id] {
+			noReceiveNodes[id] = true
+			added = append(added, id)
+		}
+	}
+	for id, disks := range ecb.rejectedDisks {
+		if noReceiveNodes[id] {
+			continue
+		}
+		if noReceiveDisks[id] == nil {
+			noReceiveDisks[id] = make(map[uint32]bool)
+		}
+		for diskId := range disks {
+			if !noReceiveDisks[id][diskId] {
+				noReceiveDisks[id][diskId] = true
+				added = append(added, fmt.Sprintf("%s:disk%d", id, diskId))
+			}
+		}
+	}
+	sort.Strings(added)
+	return
+}
+
+func (ecb *ecBalancer) hasOtherError() bool {
+	ecb.mu.Lock()
+	defer ecb.mu.Unlock()
+	return ecb.sawOtherErr
+}
+
+// recordMoveResult journals an apply-mode move outcome: a success is replayed
+// onto the next attempt's snapshot, a "no space left" marks the target
+// no-receive, and any other failure makes the run non-retryable.
+func (ecb *ecBalancer) recordMoveResult(m ecbalancer.Move, err error) {
+	ecb.mu.Lock()
+	defer ecb.mu.Unlock()
+	switch {
+	case err == nil:
+		ecb.completedMoves = append(ecb.completedMoves, m)
+	case m.TargetNode != "" && volume_move.IsNoSpaceLeftError(err):
+		if m.TargetDisk > 0 {
+			// The refusal convicts only the named disk; the node's other
+			// disks may still have room.
+			if ecb.rejectedDisks == nil {
+				ecb.rejectedDisks = make(map[string]map[uint32]bool)
+			}
+			if ecb.rejectedDisks[m.TargetNode] == nil {
+				ecb.rejectedDisks[m.TargetNode] = make(map[uint32]bool)
+			}
+			ecb.rejectedDisks[m.TargetNode][m.TargetDisk] = true
+		} else {
+			// The server auto-picked and found no usable disk: the whole
+			// node is out.
+			if ecb.rejectedTargets == nil {
+				ecb.rejectedTargets = make(map[string]bool)
+			}
+			ecb.rejectedTargets[m.TargetNode] = true
+		}
+	default:
+		ecb.sawOtherErr = true
 	}
-	return ecb.balance(collections)
 }
 
 // defaultECRatio resolves a collection's EC data/parity counts, defaulting to
@@ -852,7 +1102,7 @@ func defaultECRatio(_ string) (int, int) {
 // balance plans EC shard moves with the shared planner and executes them. When
 // collections is empty all collections present are balanced.
 func (ecb *ecBalancer) balance(collections []string) error {
-	topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds)
+	topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds, ecb.noReceiveTargets, ecb.noReceiveDisks)
 	if len(ecb.volumeIds) > 0 {
 		requested := make([]uint32, 0, len(ecb.volumeIds))
 		for vid := range ecb.volumeIds {
@@ -896,6 +1146,7 @@ func (ecb *ecBalancer) balance(collections []string) error {
 		}
 		fmt.Printf("planned %d ec shard move(s) and %d ec shard deletion(s)\n", len(moves)-deletions, deletions)
 	}
+	ecb.plannedMoves = len(moves)
 	return ecb.executeMoves(moves)
 }
 
@@ -907,7 +1158,7 @@ func (ecb *ecBalancer) balance(collections []string) error {
 // (0,0 when unreported, e.g. always in OSS), which Plan prefers over the
 // collection ratio for mixed-ratio clusters, and the set of volume ids that made
 // it into the topology.
-func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool, migratingVolumeIds map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) {
+func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool, migratingVolumeIds map[uint32]bool, noReceiveTargets map[string]bool, noReceiveDisks map[string]map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) {
 	allowed := make(map[string]bool, len(collections))
 	for _, c := range collections {
 		allowed[c] = true
@@ -924,11 +1175,17 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.
 	for _, en := range ecNodes {
 		rackKey := string(en.DC) + ":" + string(en.Rack)
 		node := topo.AddNode(en.Info.Id, string(en.DC), rackKey, en.FreeEcSlot)
+		if noReceiveTargets[en.Info.Id] {
+			node.SetNoReceive()
+		}
 		// Group by physical machine (host) so shards spread across machines, not just
 		// nodes; the id stays the node identity used for moves.
 		node.SetHost(pb.NewServerAddressFromDataNode(en.Info).ToHost())
 		for diskId, d := range en.Disks {
 			node.AddDisk(diskId, d.DiskType, d.FreeEcSlots, d.EcShardCount)
+			if noReceiveDisks[en.Info.Id][diskId] {
+				node.SetDiskNoReceive(diskId)
+			}
 		}
 		for diskTypeKey, diskInfo := range en.Info.DiskInfos {
 			if diskInfo == nil {
@@ -1069,14 +1326,16 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e
 		// thing making this safe -- and the plan saying so is not evidence. A
 		// topology entry can name a location holding nothing, and deleting on
 		// that basis removes the last copy. Confirm the keep node has it.
-		if err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId); err != nil {
-			return err
-		}
+		err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId)
 		addr := pb.NewServerAddressFromDataNode(src.Info)
-		if err := UnmountEcShards(grpcDialOption, vid, addr, shardIds); err != nil {
-			return err
+		if err == nil {
+			err = UnmountEcShards(grpcDialOption, vid, addr, shardIds)
 		}
-		return SourceServerDeleteEcShards(grpcDialOption, m.Collection, vid, addr, shardIds)
+		if err == nil {
+			err = SourceServerDeleteEcShards(grpcDialOption, m.Collection, vid, addr, shardIds)
+		}
+		ecb.recordMoveResult(m, err)
+		return err
 	}
 
 	dst := byID[m.TargetNode]
@@ -1092,7 +1351,9 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e
 		// Dry-run: update the in-memory model only.
 		return MoveMountedShardToEcNode(ecb.env, src, m.Collection, vid, shardId, dst, m.TargetDisk, false, ecb.diskType)
 	}
-	return ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk)
+	err := ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk)
+	ecb.recordMoveResult(m, err)
+	return err
 }
 
 // applyShardMoveRPC copies a shard to the destination disk, verifies the

weed/filer/abstract_sql/abstract_sql_store.go

@@ -390,6 +390,14 @@ func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpat
 	return doDeleteFolderChildren()
 }
 
+var likeEscaper = strings.NewReplacer("!", "!!", "%", "!%", "_", "!_")
+
+// LikePattern escapes LIKE wildcards so the prefix matches literally;
+// '!' pairs with the ESCAPE '!' clause emitted by the SqlGenerators.
+func LikePattern(prefix string) string {
+	return likeEscaper.Replace(prefix) + "%"
+}
+
 func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 
 	db, bucket, shortPath, err := store.getTxOrDB(ctx, dirPath, true)
@@ -402,7 +410,7 @@ func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context,
 		sqlText = store.GetSqlListInclusive(bucket)
 	}
 
-	rows, err := db.QueryContext(ctx, sqlText, util.HashStringToLong(string(shortPath)), startFileName, string(shortPath), prefix+"%", limit+1)
+	rows, err := db.QueryContext(ctx, sqlText, util.HashStringToLong(string(shortPath)), startFileName, string(shortPath), LikePattern(prefix), limit+1)
 	if err != nil {
 		return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
 	}

weed/filer/abstract_sql/abstract_sql_store_test.go

@@ -52,6 +52,24 @@ func TestGetTxOrDBRealBucket(t *testing.T) {
 	}
 }
 
+func TestLikePattern(t *testing.T) {
+	cases := []struct {
+		prefix, want string
+	}{
+		{"", "%"},
+		{"plain", "plain%"},
+		{"call_03_KNl", "call!_03!_KNl%"},
+		{"100%", "100!%%"},
+		{"a!b", "a!!b%"},
+		{`a\b`, `a\b%`},
+	}
+	for _, c := range cases {
+		if got := LikePattern(c.prefix); got != c.want {
+			t.Errorf("LikePattern(%q) = %q, want %q", c.prefix, got, c.want)
+		}
+	}
+}
+
 func TestSplitPoolForKv(t *testing.T) {
 	cases := []struct {
 		maxOpen, mainOpen, kvOpen int

weed/filer/filer.go

@@ -890,7 +890,7 @@ func (f *Filer) DeleteEmptyParentDirectories(ctx context.Context, dirPath util.F
 	glog.V(2).InfofCtx(ctx, "DeleteEmptyParentDirectories: deleting empty directory %s", dirPath)
 	parentDir, _ := dirPath.DirAndName()
 	if dirEntry, findErr := f.FindEntry(ctx, dirPath); findErr == nil {
-		if delErr := f.doDeleteEntryMetaAndData(ctx, dirEntry, false, false, nil); delErr == nil {
+		if delErr := f.doDeleteEntryMetaAndData(f.emptyFolderDeleteContext(ctx, dirPath), dirEntry, false, false, nil); delErr == nil {
 			// Successfully deleted, continue checking upwards
 			f.DeleteEmptyParentDirectories(ctx, util.FullPath(parentDir), stopAtPath)
 		} else {

weed/filer/filer_delete_entry.go

@@ -80,6 +80,27 @@ func keepsRemoteObject(ctx context.Context) bool {
 	return ctx.Value(keepRemoteObjectKey{}) != nil
 }
 
+// emptyFolderDeleteContext makes deleting a folder that is empty in the filer
+// metadata-only when it is under a remote mount. A mount is populated lazily,
+// so the folder can still hold remote objects never read, and removing its
+// remote directory deletes every object under that prefix.
+func (f *Filer) emptyFolderDeleteContext(ctx context.Context, dirPath util.FullPath) context.Context {
+	if f.IsUnderRemoteMount(dirPath) {
+		return WithKeepRemoteObject(ctx)
+	}
+	return ctx
+}
+
+// IsUnderRemoteMount reports whether p is a remote mount directory or lies under one.
+func (f *Filer) IsUnderRemoteMount(p util.FullPath) bool {
+	if f.RemoteStorage == nil {
+		return false
+	}
+	// Mount rules are keyed with a trailing slash, so this also matches the mount directory itself.
+	_, remoteLoc := f.RemoteStorage.FindMountDirectory(util.FullPath(string(p) + "/"))
+	return remoteLoc != nil
+}
+
 func deleteEventEntry(ctx context.Context, entry *Entry) *Entry {
 	if !keepsRemoteObject(ctx) {
 		return entry
@@ -108,6 +129,9 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 	if ifNotModifiedAfter > 0 && entry.Attr.Mtime.Unix() > ifNotModifiedAfter {
 		return nil
 	}
+	if !isRecursive && entry.IsDirectory() {
+		ctx = f.emptyFolderDeleteContext(ctx, p)
+	}
 	isDeleteCollection := f.IsBucket(entry)
 	collectionName := ""
 	if isDeleteCollection {
@@ -170,7 +194,7 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks, isDeletingBucket, isFromOtherCluster bool, signatures []int32, onHardLinkIdsFn OnHardLinkIdsFunc) (err error) {
 
 	var dirTombstoneTs int64
-	if isRecursive {
+	if isRecursive && !keepsRemoteObject(ctx) {
 		// Tombstone the directory before its children: when the store drops
 		// the subtree without listing it, or a child error aborts the walk,
 		// the ancestor tombstone still covers every descendant.
@@ -201,7 +225,9 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 
 			for _, sub := range entries {
 				lastFileName = sub.Name()
-				f.noteRemoteDeletion(sub.FullPath, sub.IsDirectory(), time.Now().UnixNano())
+				if !keepsRemoteObject(ctx) {
+					f.noteRemoteDeletion(sub.FullPath, sub.IsDirectory(), time.Now().UnixNano())
+				}
 				if sub.IsDirectory() {
 					subIsDeletingBucket := f.IsBucket(sub)
 					err = f.doBatchDeleteFolderMetaAndData(ctx, sub, isRecursive, ignoreRecursiveError, shouldDeleteChunks, subIsDeletingBucket, isFromOtherCluster, nil, onHardLinkIdsFn)
@@ -264,7 +290,10 @@ func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shou
 		}
 	}
 
-	f.noteRemoteDeletion(entry.FullPath, entry.IsDirectory(), time.Now().UnixNano())
+	// A metadata-only delete owes the remote nothing, so lazy reads stay open.
+	if !keepsRemoteObject(ctx) {
+		f.noteRemoteDeletion(entry.FullPath, entry.IsDirectory(), time.Now().UnixNano())
+	}
 
 	if storeDeletionErr := f.Store.DeleteOneEntry(ctx, entry); storeDeletionErr != nil {
 		return fmt.Errorf("filer store delete: %w", storeDeletionErr)

weed/filer/filer_lazy_remote.go

@@ -18,11 +18,15 @@ import (
 
 type lazyFetchContextKey struct{}
 
+// lazyParentsContextKey marks a FindEntry issued while materializing missing
+// parents: that fetch's own parent walk would only re-stat the same gaps.
+type lazyParentsContextKey struct{}
+
 // maybeLazyFetchFromRemote is called by FindEntry when the store returns no
 // entry for p. If p is under a remote-storage mount, it stats the remote
 // object, builds a filer Entry from the result, and persists it via
-// CreateEntry with SkipCheckParentDirectory so phantom parent directories
-// under the mount are not required.
+// CreateEntry together with its missing parent directories: remote.uncache
+// and cache eviction find entries by walking down from the mount directory.
 //
 // On a CreateEntry failure after a successful StatFile the in-memory entry is
 // still returned (availability over consistency); the singleflight key is
@@ -127,7 +131,10 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) (
 			glog.V(2).InfofCtx(ctx, "maybeLazyFetchFromRemote: %s deleted during remote stat", p)
 			return lazyFetchResult{nil}, nil
 		}
-		saveErr := f.CreateEntry(persistCtx, entry, nil, false, false, nil, true, f.MaxFilenameLength)
+		if ctx.Value(lazyParentsContextKey{}) == nil {
+			f.fetchMissingRemoteParents(persistBaseCtx, mountDir, p)
+		}
+		saveErr := f.CreateEntry(persistCtx, entry, nil, false, false, nil, false, f.MaxFilenameLength)
 		if saveErr != nil {
 			glog.Warningf("maybeLazyFetchFromRemote: failed to persist filer entry for %s: %v", p, saveErr)
 			f.lazyFetchGroup.Forget(key)
@@ -157,6 +164,25 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) (
 	return result.entry, nil
 }
 
+// fetchMissingRemoteParents looks up p's missing parents under mountDir, top
+// down, through lazy fetch. A remote object stored at a parent's key is then
+// materialized first, so creating p promotes it to a prefix object instead of
+// hiding it behind a plain directory. Each missing parent costs one remote
+// stat, once per directory.
+func (f *Filer) fetchMissingRemoteParents(ctx context.Context, mountDir, p util.FullPath) {
+	var missing []util.FullPath
+	for dir, _ := p.DirAndName(); len(dir) > len(mountDir); dir, _ = util.FullPath(dir).DirAndName() {
+		if _, err := f.Store.FindEntry(ctx, util.FullPath(dir)); !errors.Is(err, filer_pb.ErrNotFound) {
+			break
+		}
+		missing = append(missing, util.FullPath(dir))
+	}
+	parentCtx := context.WithValue(ctx, lazyParentsContextKey{}, true)
+	for i := len(missing) - 1; i >= 0; i-- {
+		f.FindEntry(parentCtx, missing[i])
+	}
+}
+
 // retractLazyRemoteEntry deletes the entry at entry.FullPath only when it is
 // still the entry a lazy remote read just materialized — a concurrent write
 // may have replaced it, and deleting by path alone would take that write down.

weed/filer/filer_lazy_remote_listing.go

@@ -103,6 +103,10 @@ func (f *Filer) maybeLazyListFromRemote(ctx context.Context, p util.FullPath) {
 		persistCtx := context.WithValue(context.WithoutCancel(ctx), lazyListContextKey{}, true)
 		persistCtx = context.WithValue(persistCtx, lazyFetchContextKey{}, true)
 
+		// Entries listed under p must stay reachable by a walk from the mount
+		// directory, the same reason the lazy fetch fills in missing parents.
+		f.fetchMissingRemoteParents(context.WithoutCancel(ctx), mountDir, p)
+
 		listErr := client.ListDirectory(persistCtx, objectLoc, func(dir string, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error {
 			childPath := p.Child(name)
 
@@ -215,7 +219,7 @@ func (f *Filer) updateDirectoryListingSyncedAt(ctx context.Context, p util.FullP
 			dirEntry.Extended = make(map[string][]byte)
 		}
 		dirEntry.Extended[xattrRemoteListingSyncedAt] = []byte(fmt.Sprintf("%d", syncTime.Unix()))
-		if saveErr := f.CreateEntry(ctx, dirEntry, nil, false, false, nil, true, f.MaxFilenameLength); saveErr != nil {
+		if saveErr := f.CreateEntry(ctx, dirEntry, nil, false, false, nil, false, f.MaxFilenameLength); saveErr != nil {
 			glog.Warningf("maybeLazyListFromRemote: create dir synced_at for %s: %v", p, saveErr)
 		}
 		return

weed/filer/filer_lazy_remote_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
@@ -35,6 +36,7 @@ type stubFilerStore struct {
 	kv              map[string][]byte
 	insertErr       error
 	findErr         error
+	findErrByPath   map[string]error
 	kvPutErr        error
 	kvGetErr        error
 	kvDeleteErr     error
@@ -200,6 +202,9 @@ func (s *stubFilerStore) FindEntry(_ context.Context, p util.FullPath) (*Entry,
 	if s.findErr != nil {
 		return nil, s.findErr
 	}
+	if err := s.findErrByPath[string(p)]; err != nil {
+		return nil, err
+	}
 	if e, ok := s.entries[string(p)]; ok {
 		return e, nil
 	}
@@ -340,6 +345,11 @@ func putConfigEntry(store *stubFilerStore, path string, content []byte) {
 	}
 }
 
+// putDirEntry stores a plain directory entry, so a lazy fetch below it finds its parent in place.
+func putDirEntry(store *stubFilerStore, path string) {
+	store.entries[path] = &Entry{FullPath: util.FullPath(path), Attr: Attr{Mode: os.ModeDir | 0755}}
+}
+
 // --- tests ---
 
 func TestMaybeLazyFetchFromRemote_HitsRemoteAndPersists(t *testing.T) {
@@ -375,6 +385,199 @@ func TestMaybeLazyFetchFromRemote_HitsRemoteAndPersists(t *testing.T) {
 	assert.Equal(t, []byte("zstd"), stored.Extended["Content-Encoding"])
 }
 
+// pathStatRemoteClient stats only the objects it holds, keyed by remote path.
+type pathStatRemoteClient struct {
+	stubRemoteClient
+	objects   map[string]*filer_pb.RemoteEntry
+	statCalls []string
+}
+
+func (c *pathStatRemoteClient) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) {
+	c.statCalls = append(c.statCalls, loc.Path)
+	if remoteEntry, found := c.objects[loc.Path]; found {
+		return remoteEntry, nil
+	}
+	return nil, remote_storage.ErrRemoteObjectNotFound
+}
+
+func newPathStatFiler(t *testing.T, storageType string, client remote_storage.RemoteStorageClient) (*Filer, *stubFilerStore) {
+	t.Helper()
+	t.Cleanup(registerStubMaker(t, storageType, client))
+	conf := &remote_pb.RemoteConf{Name: "mystore", Type: storageType}
+	rs := NewFilerRemoteStorage()
+	rs.storageNameToConf[conf.Name] = conf
+	rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{
+		Name:   "mystore",
+		Bucket: "mybucket",
+		Path:   "/",
+	})
+	store := newStubFilerStore()
+	return newTestFiler(t, store, rs), store
+}
+
+func TestMaybeLazyFetchFromRemote_PersistsParentDirectories(t *testing.T) {
+	stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
+		"/a/b/file.bin":  {RemoteMtime: 1700000000, RemoteSize: 1234},
+		"/a/b/other.bin": {RemoteMtime: 1700000000, RemoteSize: 5678},
+	}}
+	f, store := newPathStatFiler(t, "stub_lazy_parents", stub)
+
+	const filePath = "/buckets/mybucket/a/b/file.bin"
+	entry, err := f.maybeLazyFetchFromRemote(context.Background(), filePath)
+	require.NoError(t, err)
+	require.NotNil(t, entry)
+	assert.Equal(t, []string{"/a/b/file.bin", "/a", "/a/b"}, stub.statCalls, "each missing parent is stat'ed once")
+
+	for _, dir := range []string{"/buckets/mybucket/a", "/buckets/mybucket/a/b"} {
+		dirEntry := store.getEntry(dir)
+		require.NotNil(t, dirEntry, "parent %s not persisted", dir)
+		assert.True(t, dirEntry.IsDirectory(), "%s should be a directory", dir)
+		assert.Nil(t, dirEntry.Remote, "%s holds no remote object", dir)
+	}
+
+	// A sibling finds its parents in place and stats only itself.
+	stub.statCalls = nil
+	_, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/a/b/other.bin")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"/a/b/other.bin"}, stub.statCalls)
+
+	// Once the content is cached, eviction walks down from the mount and must reach it.
+	stored := store.getEntry(filePath)
+	require.NotNil(t, stored)
+	stored.Chunks = []*filer_pb.FileChunk{{FileId: "1,01637037d6", Size: 1234}}
+	stored.Remote.LastLocalSyncTsNs = time.Now().UnixNano()
+	require.NoError(t, store.UpdateEntry(context.Background(), stored))
+
+	evictable := f.ListEvictableRemoteEntries(context.Background(), []util.FullPath{"/buckets/mybucket"}, 0)
+	require.Len(t, evictable, 1)
+	assert.Equal(t, util.FullPath(filePath), evictable[0].FullPath)
+}
+
+func TestMaybeLazyFetchFromRemote_ParentLookupStopsAtStoreError(t *testing.T) {
+	stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
+		"/a/b/file.bin": {RemoteMtime: 1700000000, RemoteSize: 1234},
+	}}
+	f, store := newPathStatFiler(t, "stub_lazy_parent_store_error", stub)
+	store.findErrByPath = map[string]error{"/buckets/mybucket/a/b": errors.New("store unavailable")}
+
+	_, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/a/b/file.bin")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"/a/b/file.bin"}, stub.statCalls, "a parent the store cannot read is not treated as missing")
+}
+
+func TestMaybeLazyFetchFromRemote_PromotesRemoteObjectAtParentKey(t *testing.T) {
+	stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
+		"/reports":          {RemoteMtime: 1700000000, RemoteSize: 10},
+		"/reports/2026.csv": {RemoteMtime: 1700000000, RemoteSize: 20},
+	}}
+	f, store := newPathStatFiler(t, "stub_lazy_prefix_object", stub)
+
+	_, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/reports/2026.csv")
+	require.NoError(t, err)
+
+	parent := store.getEntry("/buckets/mybucket/reports")
+	require.NotNil(t, parent)
+	assert.True(t, parent.IsDirectory())
+	assert.Equal(t, []byte("true"), parent.Extended[s3_constants.SeaweedFSPrefixObject], "the remote object at the parent key must stay readable")
+	require.NotNil(t, parent.Remote)
+	assert.Equal(t, int64(10), parent.Remote.RemoteSize)
+}
+
+func TestMaybeLazyFetchFromRemote_StatsEachParentOnce(t *testing.T) {
+	stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
+		"/a/b/c/file.bin": {RemoteMtime: 1700000000, RemoteSize: 1234},
+	}}
+	f, _ := newPathStatFiler(t, "stub_lazy_parent_stats", stub)
+
+	_, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/a/b/c/file.bin")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"/a/b/c/file.bin", "/a", "/a/b", "/a/b/c"}, stub.statCalls,
+		"a parent still missing after its own stat must not be re-stat'ed by the next lookup")
+}
+
+func TestMaybeLazyListFromRemote_PersistsParentDirectories(t *testing.T) {
+	stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{}}
+	stub.listDirFn = func(loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) error {
+		return visitFn("/", "file.bin", false, &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 1234})
+	}
+	f, store := newPathStatFiler(t, "stub_lazy_list_parents", stub)
+	f.RemoteStorage.mapDirectoryToRemoteStorage("/buckets/listmount", &remote_pb.RemoteStorageLocation{
+		Name:                   "mystore",
+		Bucket:                 "listmount",
+		Path:                   "/",
+		ListingCacheTtlSeconds: 300,
+	})
+
+	f.maybeLazyListFromRemote(context.Background(), util.FullPath("/buckets/listmount/a/b"))
+
+	assert.Equal(t, []string{"/a"}, stub.statCalls, "each missing parent is stat'ed once")
+	for _, dir := range []string{"/buckets/listmount", "/buckets/listmount/a", "/buckets/listmount/a/b"} {
+		dirEntry := store.getEntry(dir)
+		require.NotNil(t, dirEntry, "%s not persisted", dir)
+		assert.True(t, dirEntry.IsDirectory(), "%s should be a directory", dir)
+	}
+
+	child := store.getEntry("/buckets/listmount/a/b/file.bin")
+	require.NotNil(t, child)
+	child.Chunks = []*filer_pb.FileChunk{{FileId: "1,01637037d6", Size: 1234}}
+	child.Remote.LastLocalSyncTsNs = time.Now().UnixNano()
+	require.NoError(t, store.UpdateEntry(context.Background(), child))
+
+	evictable := f.ListEvictableRemoteEntries(context.Background(), []util.FullPath{"/buckets/listmount"}, 0)
+	require.Len(t, evictable, 1)
+	assert.Equal(t, util.FullPath("/buckets/listmount/a/b/file.bin"), evictable[0].FullPath)
+}
+
+func TestDeleteEntryMetaAndData_EmptyFolderUnderMountKeepsRemotePrefix(t *testing.T) {
+	stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
+		"/dir/unread.bin": {RemoteMtime: 1700000000, RemoteSize: 1},
+	}}
+	f, store := newPathStatFiler(t, "stub_lazy_empty_folder", stub)
+	dirPath := util.FullPath("/buckets/mybucket/dir")
+	putDirEntry(store, string(dirPath))
+
+	require.NoError(t, f.DeleteEntryMetaAndData(context.Background(), dirPath, false, false, false, false, nil, 0))
+
+	assert.Empty(t, stub.removeCalls, "an empty local folder must not remove its remote prefix")
+	assert.Nil(t, store.getEntry(string(dirPath)))
+	entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/unread.bin")
+	require.NoError(t, err)
+	assert.NotNil(t, entry, "a metadata-only delete must not block lazy reads under the folder")
+}
+
+func TestDeleteEmptyParentDirectories_UnderMountKeepsRemotePrefix(t *testing.T) {
+	stub := &pathStatRemoteClient{}
+	f, store := newPathStatFiler(t, "stub_lazy_expired_parents", stub)
+	f.DirBucketsPath = "/buckets"
+	putDirEntry(store, "/buckets/mybucket/a")
+	putDirEntry(store, "/buckets/mybucket/a/b")
+
+	f.DeleteEmptyParentDirectories(context.Background(), "/buckets/mybucket/a/b", "/buckets")
+
+	assert.Empty(t, stub.removeCalls, "an empty local folder must not remove its remote prefix")
+	assert.Nil(t, store.getEntry("/buckets/mybucket/a/b"))
+	assert.Nil(t, store.getEntry("/buckets/mybucket/a"))
+}
+
+func TestIsUnderRemoteMount(t *testing.T) {
+	rs := NewFilerRemoteStorage()
+	rs.mapDirectoryToRemoteStorage("/buckets/mirror", &remote_pb.RemoteStorageLocation{
+		Name:   "mystore",
+		Bucket: "mirror",
+		Path:   "/",
+	})
+	f := newTestFiler(t, newStubFilerStore(), rs)
+
+	for path, want := range map[util.FullPath]bool{
+		"/buckets/mirror":     true,
+		"/buckets/mirror/a/b": true,
+		"/buckets/mirror2/a":  false,
+		"/buckets/local/a":    false,
+	} {
+		assert.Equal(t, want, f.IsUnderRemoteMount(path), path)
+	}
+}
+
 func TestMaybeLazyFetchFromRemote_NotUnderMount(t *testing.T) {
 	rs := NewFilerRemoteStorage()
 	store := newStubFilerStore()

weed/filer/filer_remote_tombstone.go

@@ -407,7 +407,7 @@ func (f *Filer) onRemoteDeletionEvents(event *filer_pb.SubscribeMetadataResponse
 	if message == nil {
 		return
 	}
-	if message.OldEntry != nil {
+	if message.OldEntry != nil && !IsMetadataOnlyDelete(message.OldEntry) {
 		sourcePath := filer_pb.MetadataEventSourceFullPath(event)
 		if message.NewEntry == nil || sourcePath != filer_pb.MetadataEventTargetFullPath(event) {
 			if f.RemoteStorage != nil && f.remoteTombstones != nil {

weed/filer/filer_remote_tombstone_test.go

@@ -187,6 +187,7 @@ func TestMaybeLazyFetchFromRemote_SyncOffsetReleasesTombstone(t *testing.T) {
 		},
 	}
 	f, store := newMountedTestFiler(t, storageType, stub, 0)
+	putDirEntry(store, "/buckets/mybucket/dir")
 
 	filePath := util.FullPath("/buckets/mybucket/dir/a.txt")
 	deleteTsNs := time.Now().UnixNano()
@@ -222,6 +223,7 @@ func TestDeleteEntryMetaAndData_TombstonesPath(t *testing.T) {
 		},
 	}
 	f, store := newMountedTestFiler(t, storageType, stub, 0)
+	putDirEntry(store, "/buckets/mybucket/dir")
 
 	filePath := util.FullPath("/buckets/mybucket/dir/a.txt")
 	store.entries[string(filePath)] = &Entry{
@@ -257,6 +259,35 @@ func TestDeleteEntryMetaAndData_TombstonesPath(t *testing.T) {
 	assert.Equal(t, 1, stub.statCalls)
 }
 
+func TestOnMetadataChangeEvent_MetadataOnlyDeleteSkipsTombstone(t *testing.T) {
+	const storageType = "stub_tomb_metadata_only"
+	stub := &countingRemoteClient{
+		stubRemoteClient: stubRemoteClient{
+			statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 11},
+		},
+	}
+	f, store := newMountedTestFiler(t, storageType, stub, 0)
+	putDirEntry(store, "/buckets/mybucket/dir")
+
+	f.onMetadataChangeEvent(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/mybucket",
+		TsNs:      time.Now().UnixNano(),
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{
+				Name:        "dir",
+				IsDirectory: true,
+				Extended:    map[string][]byte{ExtKeepRemoteObjectKey: []byte("true")},
+			},
+		},
+	})
+
+	// nothing is owed to the remote, so objects under the folder stay readable
+	entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/a.txt")
+	require.NoError(t, err)
+	require.NotNil(t, entry)
+	assert.Equal(t, 1, stub.statCalls)
+}
+
 func TestOnMetadataChangeEvent_PeerDeleteTombstones(t *testing.T) {
 	const storageType = "stub_tomb_peer"
 	stub := &countingRemoteClient{

weed/filer/mysql/mysql_sql_gen.go

@@ -64,12 +64,12 @@ func (gen *SqlGenMysql) nameExpr() string {
 
 func (gen *SqlGenMysql) GetSqlListExclusive(tableName string) string {
 	name := gen.nameExpr()
-	return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s > ? AND `directory` = ? AND %s LIKE ? ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
+	return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s > ? AND `directory` = ? AND %s LIKE ? ESCAPE '!' ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
 }
 
 func (gen *SqlGenMysql) GetSqlListInclusive(tableName string) string {
 	name := gen.nameExpr()
-	return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s >= ? AND `directory` = ? AND %s LIKE ? ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
+	return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s >= ? AND `directory` = ? AND %s LIKE ? ESCAPE '!' ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
 }
 
 func (gen *SqlGenMysql) GetSqlCreateTable(tableName string) string {

weed/filer/mysql/mysql_sql_gen_test.go

@@ -65,6 +65,15 @@ func TestListSqlBinaryOrderingOnNonBinaryColumn(t *testing.T) {
 	}
 }
 
+func TestListSqlEscapesLikeWildcards(t *testing.T) {
+	gen := &SqlGenMysql{}
+	for _, got := range []string{gen.GetSqlListExclusive("filemeta"), gen.GetSqlListInclusive("filemeta")} {
+		if !strings.Contains(got, "LIKE ? ESCAPE '!'") {
+			t.Fatalf("list query must declare the escape char matching LikePattern, got: %s", got)
+		}
+	}
+}
+
 func TestIsBinaryCollation(t *testing.T) {
 	binary := []string{"", "binary", "utf8mb4_bin", "utf8mb3_bin", "latin1_bin", "UTF8MB4_BIN"}
 	for _, c := range binary {

weed/filer/postgres/postgres_sql_gen.go

@@ -77,12 +77,12 @@ func (gen *SqlGenPostgres) nameExpr() string {
 
 func (gen *SqlGenPostgres) GetSqlListExclusive(tableName string) string {
 	name := gen.nameExpr()
-	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>$2 AND directory=$3 AND %s like $4 ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
+	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>$2 AND directory=$3 AND %s like $4 ESCAPE '!' ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
 }
 
 func (gen *SqlGenPostgres) GetSqlListInclusive(tableName string) string {
 	name := gen.nameExpr()
-	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>=$2 AND directory=$3 AND %s like $4 ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
+	return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>=$2 AND directory=$3 AND %s like $4 ESCAPE '!' ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
 }
 
 func (gen *SqlGenPostgres) GetSqlCreateTable(tableName string) string {

weed/filer/postgres/postgres_sql_gen_test.go

@@ -62,6 +62,15 @@ func TestListSqlBinaryOrderingOnNonBinaryColumn(t *testing.T) {
 	}
 }
 
+func TestListSqlEscapesLikeWildcards(t *testing.T) {
+	gen := &SqlGenPostgres{}
+	for _, got := range []string{gen.GetSqlListExclusive("filemeta"), gen.GetSqlListInclusive("filemeta")} {
+		if !strings.Contains(got, "like $4 ESCAPE '!'") {
+			t.Fatalf("list query must declare the escape char matching LikePattern, got: %s", got)
+		}
+	}
+}
+
 func TestIsByteOrderedCollation(t *testing.T) {
 	byteOrdered := []string{"C", "POSIX", "c", "C.UTF-8", "C.utf8"}
 	for _, c := range byteOrdered {

weed/filer/sqlite/sqlite_like_escape_test.go

@@ -0,0 +1,75 @@
+//go:build sqlite
+
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/filer/mysql"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	_ "modernc.org/sqlite"
+)
+
+func TestListDirectoryPrefixedEntriesEscapesLikeWildcards(t *testing.T) {
+	db, err := sql.Open("sqlite", "file::memory:?cache=shared")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	_, err = db.Exec(`CREATE TABLE filemeta (dirhash BIGINT, name VARCHAR(1000), directory TEXT, meta BLOB, PRIMARY KEY (dirhash, name))`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	store := &SqliteStore{}
+	store.SqlGenerator = &mysql.SqlGenMysql{}
+	store.DB = db
+
+	ctx := context.Background()
+	names := []string{
+		"call_03_KNlHPt6grkZMe4WZVwQi9710", // target: literal _ prefix
+		"callX03XKNlZZZ",                   // would match if _ were a wildcard
+		"call_99_other",
+		"100%_done", // % and _ in the name
+		"100aXdone", // would match prefix "100%_" if unescaped
+		"bang!file",
+	}
+	for _, n := range names {
+		entry := &filer.Entry{FullPath: util.FullPath("/dir/" + n)}
+		if err := store.InsertEntry(ctx, entry); err != nil {
+			t.Fatalf("insert %s: %v", n, err)
+		}
+	}
+
+	list := func(prefix string) []string {
+		var got []string
+		_, err := store.ListDirectoryPrefixedEntries(ctx, "/dir", "", false, 100, prefix, func(e *filer.Entry) (bool, error) {
+			got = append(got, e.Name())
+			return true, nil
+		})
+		if err != nil {
+			t.Fatalf("list %q: %v", prefix, err)
+		}
+		return got
+	}
+
+	if got := list("call_03_"); len(got) != 1 || got[0] != "call_03_KNlHPt6grkZMe4WZVwQi9710" {
+		t.Fatalf("prefix with _: got %v", got)
+	}
+	if got := list("call_"); len(got) != 2 {
+		t.Fatalf("prefix call_: got %v", got)
+	}
+	if got := list("100%_"); len(got) != 1 || got[0] != "100%_done" {
+		t.Fatalf("prefix with %%: got %v", got)
+	}
+	if got := list("bang!"); len(got) != 1 || got[0] != "bang!file" {
+		t.Fatalf("prefix with !: got %v", got)
+	}
+	if got := list(""); len(got) != len(names) {
+		t.Fatalf("empty prefix: got %v", got)
+	}
+}

weed/filer/ydb/ydb_queries.go

@@ -55,7 +55,7 @@ const (
 		
 		SELECT name, meta
 		FROM ` + asql.DEFAULT_TABLE + `
-		WHERE dir_hash = $dir_hash AND directory = $directory and name > $start_name and name LIKE $prefix
+		WHERE dir_hash = $dir_hash AND directory = $directory and name > $start_name and name LIKE $prefix ESCAPE '!'
 		ORDER BY name ASC LIMIT $limit;`
 
 	listInclusiveDirectoryQuery = `
@@ -68,6 +68,6 @@ const (
 		
 		SELECT name, meta
 		FROM ` + asql.DEFAULT_TABLE + `
-		WHERE dir_hash = $dir_hash AND directory = $directory and name >= $start_name and name LIKE $prefix
+		WHERE dir_hash = $dir_hash AND directory = $directory and name >= $start_name and name LIKE $prefix ESCAPE '!'
 		ORDER BY name ASC LIMIT $limit;`
 )

weed/filer/ydb/ydb_store.go

@@ -288,7 +288,7 @@ func (store *YdbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath
 			table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(*shortDir))),
 			table.ValueParam("$directory", types.UTF8Value(*shortDir)),
 			table.ValueParam("$start_name", types.UTF8Value(startFileName)),
-			table.ValueParam("$prefix", types.UTF8Value(prefix+"%")),
+			table.ValueParam("$prefix", types.UTF8Value(abstract_sql.LikePattern(prefix))),
 			table.ValueParam("$limit", types.Uint64Value(uint64(chunkLimit))),
 		)
 

weed/operation/volume_move/ec_move.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
@@ -202,6 +203,14 @@ func (m *Mover) DeleteEcShards(ctx context.Context, volumeId needle.VolumeId, co
 	})
 }
 
+// IsNoSpaceLeftError matches the volume server's refusal when no local disk
+// can take the copied shard ("no space left" from FindEcShardTargetLocation,
+// an ENOSPC writing to a caller-pinned disk, and the Rust server's same
+// reply), however the move chain wraps it.
+func IsNoSpaceLeftError(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "no space left")
+}
+
 // RemoveEcShards unmounts then deletes shards in place — the dedup path for a
 // shard that already has a copy elsewhere.
 func (m *Mover) RemoveEcShards(ctx context.Context, volumeId needle.VolumeId, collection string, server pb.ServerAddress, shardIds []erasure_coding.ShardId) error {

weed/server/filer_server_handlers_write_autochunk.go

@@ -58,7 +58,10 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
 			writeJsonError(w, r, util.HttpStatusCancelled, err)
 		case strings.HasSuffix(errStr, "is a file") || strings.HasSuffix(errStr, "already exists"):
 			writeJsonError(w, r, http.StatusConflict, err)
-		case errStr == constants.ErrMsgBadDigest:
+		case errStr == constants.ErrMsgBadDigest,
+			errStr == constants.ErrMsgAppendSaveInsideChunked,
+			errStr == constants.ErrMsgAppendInlineMode,
+			errStr == constants.ErrMsgOffsetWriteInline:
 			writeJsonError(w, r, http.StatusBadRequest, err)
 		default:
 			writeJsonError(w, r, http.StatusInternalServerError, err)
@@ -99,8 +102,17 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
 	if so.SaveInside {
 		buf := bufPool.Get().(*bytes.Buffer)
 		buf.Reset()
-		buf.ReadFrom(part1)
-		filerResult, replyerr, _ = fs.saveMetaData(ctx, r, fileName, contentType, so, chunkSize, nil, nil, 0, buf.Bytes())
+		if _, err := buf.ReadFrom(part1); err != nil {
+			bufPool.Put(buf)
+			return nil, nil, fmt.Errorf("read input: %w", err)
+		}
+		md5bytes = util.Md5(buf.Bytes())
+		headerMd5 := r.Header.Get("Content-Md5")
+		if headerMd5 != "" && !(util.Base64Encode(md5bytes) == headerMd5 || fmt.Sprintf("%x", md5bytes) == headerMd5) {
+			bufPool.Put(buf)
+			return nil, nil, errors.New(constants.ErrMsgBadDigest)
+		}
+		filerResult, replyerr, _ = fs.saveMetaData(ctx, r, fileName, contentType, so, chunkSize, md5bytes, nil, int64(buf.Len()), buf.Bytes())
 		bufPool.Put(buf)
 		return
 	}
@@ -293,7 +305,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	if entry != nil {
 		inlineCandidate := len(entry.Content) > 0 ||
 			(isAppend && entry.FileSize == 0 && entry.Remote == nil && len(entry.HardLinkId) == 0 && len(entry.GetChunks()) == 0)
-		if isAppend && !so.SaveInside && content != nil && !inlineCandidate {
+		if isAppend && content != nil && !inlineCandidate {
+			if so.SaveInside {
+				replyerr = errors.New(constants.ErrMsgAppendSaveInsideChunked)
+				return
+			}
 			replyerr = fmt.Errorf("inline file changed while preparing append; retry the request")
 			return
 		}
@@ -301,11 +317,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 		if inlineCandidate {
 			if isOffsetWrite {
 				// TODO: support inline offset writes separately from append semantics.
-				replyerr = fmt.Errorf("offset write to inline small file is not supported yet")
+				replyerr = errors.New(constants.ErrMsgOffsetWriteInline)
 				return
 			}
 			if !isAppend || so.SaveInside || entry.Remote != nil || len(entry.HardLinkId) != 0 || len(entry.GetChunks()) != 0 {
-				replyerr = fmt.Errorf("append to inline content with this storage mode is not supported")
+				replyerr = errors.New(constants.ErrMsgAppendInlineMode)
 				return
 			}
 			if entry.FileSize > uint64(len(entry.Content)) {

weed/server/filer_server_handlers_write_autochunk_inline_append_test.go

@@ -270,6 +270,164 @@ func TestFilerInlineAppendMultipartPostStaysInline(t *testing.T) {
 	}
 }
 
+func TestFilerSaveInsideMultipartPostRecordsSize(t *testing.T) {
+	ctx := context.Background()
+	content := []byte(`{"identities":[]}`)
+	store := newRenameTestStore()
+	f := newRenameTestFiler(t, store)
+	fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+
+	var body bytes.Buffer
+	multipartWriter := multipart.NewWriter(&body)
+	part, err := multipartWriter.CreateFormFile("file", "identity.json")
+	if err != nil {
+		t.Fatalf("create multipart file part: %v", err)
+	}
+	if _, err = part.Write(content); err != nil {
+		t.Fatalf("write multipart file part: %v", err)
+	}
+	if err = multipartWriter.Close(); err != nil {
+		t.Fatalf("close multipart body: %v", err)
+	}
+	r := httptest.NewRequest(http.MethodPost, "/", &body)
+	r.Header.Set("Content-Type", multipartWriter.FormDataContentType())
+	result, requestMD5, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, int64(body.Len()), &operation.StorageOption{SaveInside: true})
+	if err != nil {
+		t.Fatalf("POST saveInside: %v", err)
+	}
+	if result == nil || result.Size != int64(len(content)) {
+		t.Fatalf("result = %#v, want size %d", result, len(content))
+	}
+	wantMD5 := md5.Sum(content)
+	if !bytes.Equal(requestMD5, wantMD5[:]) {
+		t.Fatalf("request MD5 = %x, want %x", requestMD5, wantMD5)
+	}
+	saved, err := f.FindEntry(ctx, "/identity.json")
+	if err != nil {
+		t.Fatalf("find saved entry: %v", err)
+	}
+	if !bytes.Equal(saved.Content, content) || saved.FileSize != uint64(len(content)) {
+		t.Fatalf("saved entry: content=%q size=%d, want %q size %d", saved.Content, saved.FileSize, content, len(content))
+	}
+	if !bytes.Equal(saved.Md5, wantMD5[:]) {
+		t.Fatalf("saved MD5 = %x, want %x", saved.Md5, wantMD5)
+	}
+}
+
+func saveInsideMultipartRequest(t *testing.T, target, fileName string, content []byte) *http.Request {
+	t.Helper()
+	var body bytes.Buffer
+	multipartWriter := multipart.NewWriter(&body)
+	part, err := multipartWriter.CreateFormFile("file", fileName)
+	if err != nil {
+		t.Fatalf("create multipart file part: %v", err)
+	}
+	if _, err = part.Write(content); err != nil {
+		t.Fatalf("write multipart file part: %v", err)
+	}
+	if err = multipartWriter.Close(); err != nil {
+		t.Fatalf("close multipart body: %v", err)
+	}
+	r := httptest.NewRequest(http.MethodPost, target, &body)
+	r.Header.Set("Content-Type", multipartWriter.FormDataContentType())
+	return r
+}
+
+func TestFilerSaveInsideMultipartPostRejectsBadDigest(t *testing.T) {
+	ctx := context.Background()
+	content := []byte(`{"identities":[]}`)
+	store := newRenameTestStore()
+	f := newRenameTestFiler(t, store)
+	fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+
+	r := saveInsideMultipartRequest(t, "/", "identity.json", content)
+	r.Header.Set("Content-Md5", util.Base64Encode([]byte("not the digest")))
+	if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil || !strings.Contains(err.Error(), "Content-Md5") {
+		t.Fatalf("POST saveInside error = %v, want bad Content-MD5 error", err)
+	}
+	if _, err := f.FindEntry(ctx, "/identity.json"); !errors.Is(err, filer_pb.ErrNotFound) {
+		t.Fatalf("bad digest still saved the entry: %v", err)
+	}
+}
+
+func TestFilerSaveInsideMultipartAppendRejected(t *testing.T) {
+	ctx := context.Background()
+	const filePath = "/append-target.txt"
+	appendContent := []byte("new")
+
+	t.Run("chunk backed target", func(t *testing.T) {
+		oldChunk := &filer_pb.FileChunk{FileId: "1,00000001", Offset: 0, Size: 4}
+		store := newRenameTestStore()
+		f := newRenameTestFiler(t, store)
+		if err := store.InsertEntry(ctx, &filer.Entry{
+			FullPath: util.FullPath(filePath),
+			Attr: filer.Attr{
+				Mtime:    time.Unix(1, 0),
+				Crtime:   time.Unix(1, 0),
+				FileSize: 4,
+			},
+			Chunks: []*filer_pb.FileChunk{oldChunk},
+		}); err != nil {
+			t.Fatalf("insert chunk-backed entry: %v", err)
+		}
+
+		fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+		r := saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil {
+			t.Fatal("saveInside append to chunk-backed entry succeeded")
+		}
+		recorder := httptest.NewRecorder()
+		r = saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		fs.autoChunk(ctx, recorder, r, r.ContentLength, &operation.StorageOption{SaveInside: true})
+		if recorder.Code != http.StatusBadRequest {
+			t.Fatalf("rejected saveInside append status = %d, want %d", recorder.Code, http.StatusBadRequest)
+		}
+		unchanged, err := f.FindEntry(ctx, util.FullPath(filePath))
+		if err != nil {
+			t.Fatalf("find entry after rejected append: %v", err)
+		}
+		if unchanged.FileSize != 4 || len(unchanged.Chunks) != 1 || len(unchanged.Content) != 0 {
+			t.Fatalf("rejected append changed entry: size=%d chunks=%d content=%q", unchanged.FileSize, len(unchanged.Chunks), unchanged.Content)
+		}
+	})
+
+	t.Run("inline target", func(t *testing.T) {
+		oldContent := []byte("old-")
+		store := newRenameTestStore()
+		f := newRenameTestFiler(t, store)
+		if err := store.InsertEntry(ctx, &filer.Entry{
+			FullPath: util.FullPath(filePath),
+			Attr: filer.Attr{
+				Mtime:    time.Unix(1, 0),
+				Crtime:   time.Unix(1, 0),
+				FileSize: uint64(len(oldContent)),
+			},
+			Content: append([]byte(nil), oldContent...),
+		}); err != nil {
+			t.Fatalf("insert inline entry: %v", err)
+		}
+
+		fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+		r := saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil {
+			t.Fatal("saveInside append to inline entry succeeded")
+		}
+		recorder := httptest.NewRecorder()
+		r = saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		fs.autoChunk(ctx, recorder, r, r.ContentLength, &operation.StorageOption{SaveInside: true})
+		if recorder.Code != http.StatusBadRequest {
+			t.Fatalf("rejected saveInside append status = %d, want %d", recorder.Code, http.StatusBadRequest)
+		}
+		unchanged, err := f.FindEntry(ctx, util.FullPath(filePath))
+		if err != nil {
+			t.Fatalf("find entry after rejected append: %v", err)
+		}
+		if !bytes.Equal(unchanged.Content, oldContent) || unchanged.FileSize != uint64(len(oldContent)) {
+			t.Fatalf("rejected append changed entry: content=%q size=%d", unchanged.Content, unchanged.FileSize)
+		}
+	})
+}
+
 func TestFilerInlineOffsetWriteRemainsUnsupported(t *testing.T) {
 	ctx := context.Background()
 	const filePath = "/offset.txt"

weed/server/volume_grpc_erasure_coding.go

@@ -394,7 +394,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 		}
 		location = vs.store.FindEcShardTargetLocation(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.DataShardsCount, shardIds...)
 		if location == nil {
-			return nil, fmt.Errorf("no space left")
+			return nil, status.Error(codes.ResourceExhausted, "no space left")
 		}
 	}
 

weed/server/webdav_server.go

@@ -570,6 +570,14 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
 	if fileSize == 0 {
 		return 0, io.EOF
 	}
+	if len(f.entry.Content) > 0 {
+		if f.off >= int64(len(f.entry.Content)) {
+			return 0, io.EOF
+		}
+		readSize = copy(p, f.entry.Content[f.off:])
+		f.off += int64(readSize)
+		return readSize, nil
+	}
 	if f.visibleIntervals == nil {
 		f.visibleIntervals, err = filer.NonOverlappingVisibleIntervals(f.ctx, f.fs.filerClient.GetLookupFileIdFunction(), f.entry.GetChunks(), 0, fileSize)
 		if err != nil {

weed/server/webdav_server_test.go

@@ -3,6 +3,7 @@ package weed_server
 import (
 	"context"
 	"fmt"
+	"io"
 	"net"
 	"os"
 	"testing"
@@ -104,3 +105,28 @@ func TestWebDavFile_Read_ManifestResolveFailure(t *testing.T) {
 	assert.Error(t, err)
 	assert.Equal(t, 0, n)
 }
+
+// Small files stored inline in the entry have no chunks to read from.
+func TestWebDavFile_Read_InlineContent(t *testing.T) {
+	content := []byte("hello inline")
+	entry := &filer_pb.Entry{
+		Name:       "small.txt",
+		Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))},
+		Content:    content,
+	}
+	filerAddr := startFakeWebDavFiler(t, &noVolumeFiler{})
+	fc := wdclient.NewFilerClient([]pb.ServerAddress{filerAddr}, grpc.WithTransportCredentials(insecure.NewCredentials()), "")
+	t.Cleanup(fc.Close)
+	f := &WebDavFile{fs: &WebDavFileSystem{filerClient: fc}, name: "/small.txt", entry: entry, ctx: context.Background()}
+
+	got, err := io.ReadAll(f)
+	assert.NoError(t, err)
+	assert.Equal(t, content, got)
+
+	_, err = f.Seek(6, io.SeekStart)
+	assert.NoError(t, err)
+	buf := make([]byte, 3)
+	n, err := f.Read(buf)
+	assert.NoError(t, err)
+	assert.Equal(t, "inl", string(buf[:n]))
+}

weed/shell/command_volume_server_evacuate.go

@@ -1,6 +1,7 @@
 package shell
 
 import (
+	"cmp"
 	"flag"
 	"fmt"
 	"io"
@@ -273,7 +274,7 @@ func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][
 	}
 	// most empty one is in the front
 	slices.SortFunc(otherNodes, func(a, b *Node) int {
-		return int(a.localVolumeRatio(maxVolumeCountFn) - b.localVolumeRatio(maxVolumeCountFn))
+		return cmp.Compare(a.localVolumeRatio(maxVolumeCountFn), b.localVolumeRatio(maxVolumeCountFn))
 	})
 	for i := 0; i < len(otherNodes); i++ {
 		emptyNode := otherNodes[i]

weed/shell/command_volume_server_evacuate_test.go

@@ -5,6 +5,22 @@ import (
 	"testing"
 )
 
+func TestEvacuatePrefersLowerVolumeRatio(t *testing.T) {
+	volumes := mkByteVolumes(1, 1, 1)
+	source := makeByteNode("source", 10, 0, 0, volumes)
+	fuller := makeByteNode("fuller", 10, 0, 0, mkByteVolumes(10, 4, 1))
+	emptier := makeByteNode("emptier", 10, 0, 0, mkByteVolumes(20, 2, 1))
+
+	// Put the 40%-full server first: truncating 0.2 - 0.4 to int makes the ratios tie.
+	moved, err := moveAwayOneNormalVolume(nil, nil, volumes[0], source, []*Node{fuller, emptier}, false)
+	if err != nil || !moved {
+		t.Fatalf("evacuate: moved=%v, err=%v", moved, err)
+	}
+	if _, found := emptier.selectedVolumes[volumes[0].Id]; !found {
+		t.Fatal("expected evacuation to choose the server with lower occupancy")
+	}
+}
+
 func TestVolumeServerEvacuate(t *testing.T) {
 	targetServer := ""
 	volumeRack := "rack1"

weed/storage/erasure_coding/ecbalancer/balancer.go

@@ -38,6 +38,10 @@ type Node struct {
 	rack string // composite rack key (e.g. "dc1:rack1")
 
 	freeSlots int
+	// noReceive marks a node that refused shards (e.g. answered "no space
+	// left"): it stays usable as a move source but is never picked as a target,
+	// regardless of what its freeSlots say.
+	noReceive bool
 	disks     map[uint32]*disk
 	shards    map[volKey]*volumeShards
 }
@@ -47,7 +51,8 @@ type disk struct {
 	diskType   string
 	tags       []string // placement tags, for preferred-tag tiering
 	freeSlots  int
-	shardCount int // total EC shards on this disk across all volumes
+	shardCount int  // total EC shards on this disk across all volumes
+	noReceive  bool // this disk refused a shard; siblings may still receive
 }
 
 type volumeShards struct {
@@ -156,6 +161,39 @@ func (n *Node) SetHost(host string) {
 	}
 }
 
+// SetNoReceive marks the node ineligible as a move destination. Use when the
+// node itself refused shards (e.g. answered "no space left") so the heartbeat's
+// free-slot count cannot be trusted; the node can still shed shards as a source.
+func (n *Node) SetNoReceive() {
+	n.noReceive = true
+}
+
+// SetDiskNoReceive marks one disk on the node ineligible as a move destination.
+// An explicit-disk copy can be refused while the node's other disks still have
+// room; a disk_id 0 refusal means the server searched every disk and is a
+// whole-node refusal (SetNoReceive) instead.
+func (n *Node) SetDiskNoReceive(diskID uint32) {
+	if d, ok := n.disks[diskID]; ok {
+		d.noReceive = true
+	}
+}
+
+// receiveSlots is the node's receiving capacity: its aggregate free slots with
+// refused disks subtracted, and 0 for a refused node. Slots a disk just
+// refused are phantom capacity whatever the snapshot reported.
+func (n *Node) receiveSlots() int {
+	if n.noReceive {
+		return 0
+	}
+	free := n.freeSlots
+	for _, d := range n.disks {
+		if d.noReceive {
+			free -= d.freeSlots
+		}
+	}
+	return max(free, 0)
+}
+
 // AddDisk registers a physical disk. shardCount is the disk's total EC shard
 // count across all volumes (used for disk scoring); freeSlots is the per-disk
 // free EC shard slots.
@@ -296,7 +334,7 @@ func buildRacks(nodes map[string]*Node) map[string]*rack {
 			racks[n.rack] = r
 		}
 		r.nodes[n.id] = n
-		r.freeSlots += n.freeSlots
+		r.freeSlots += n.receiveSlots()
 	}
 	return racks
 }
@@ -440,7 +478,7 @@ func planRackTotalCap(vk volKey, racks map[string]*rack, rackShardCount map[stri
 			continue
 		}
 		for _, n := range r.nodes {
-			room[rk] += min(max(n.freeSlots, 0), max(rp.SameRackCount-volumeShardCount(n, vk), 0))
+			room[rk] += min(n.receiveSlots(), max(rp.SameRackCount-volumeShardCount(n, vk), 0))
 		}
 	}
 	return rackTotalCap(sortedKeys(racks), nil, room, totalShards)
@@ -575,12 +613,15 @@ func pickBestNodeForVolumeOnDisk(nodes []*Node, vk volKey, rp *super_block.Repli
 // metadata is authoritative when present; without it the node's aggregate
 // stands in so the volume server's own disk selection can place the shard.
 func freeSlotsForType(n *Node, diskType string) int {
+	if n.noReceive {
+		return 0
+	}
 	if len(n.disks) == 0 {
 		return n.freeSlots
 	}
 	free := 0
 	for _, d := range n.disks {
-		if d.freeSlots > 0 && (diskType == "" || d.diskType == diskType) {
+		if d.freeSlots > 0 && !d.noReceive && (diskType == "" || d.diskType == diskType) {
 			free += d.freeSlots
 		}
 	}
@@ -604,7 +645,7 @@ func pickBestNodeForVolume(nodes []*Node, vk volKey, rp *super_block.ReplicaPlac
 	bestFree := -1
 	bestScore := -1
 	for _, node := range nodes {
-		if node.freeSlots <= 0 {
+		if node.receiveSlots() <= 0 {
 			continue
 		}
 		count := volumeShardCount(node, vk)
@@ -909,7 +950,7 @@ func detectGlobalImbalance(nodes map[string]*Node, racks map[string]*rack, diskT
 					continue
 				}
 				util := float64(count) / float64(capacity)
-				if util < minUtil && node.freeSlots > 0 {
+				if util < minUtil && node.receiveSlots() > 0 {
 					minUtil, minCount, minNode = util, count, node
 				}
 				if util > maxUtil {
@@ -1091,7 +1132,7 @@ func pickBestDiskOnNode(node *Node, vk volKey, diskType string, shardID, dataSha
 		if diskType != "" && d.diskType != diskType {
 			continue
 		}
-		if d.freeSlots <= 0 {
+		if d.freeSlots <= 0 || d.noReceive {
 			continue
 		}
 

weed/storage/erasure_coding/ecbalancer/balancer_test.go

@@ -926,3 +926,67 @@ func TestGlobalPrefersVolumeAbsentFromDestinationMachine(t *testing.T) {
 		t.Errorf("first global move is volume %d, want 200 (vol100 already on the destination machine)", moves[0].volumeID)
 	}
 }
+
+// TestNoReceiveNodeNeverTargeted: a node that refused a copy ("no space left")
+// must not be picked as a destination no matter how many free slots it reports,
+// but it still sheds its own over-cap shards as a source.
+func TestNoReceiveNodeNeverTargeted(t *testing.T) {
+	const vid = uint32(1)
+	topo := NewTopology()
+	src := topo.AddNode("n1", "dc1", "dc1:r1", 50)
+	src.AddShards(vid, "c", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))
+	full := topo.AddNode("n2", "dc1", "dc1:r1", 50)
+	full.AddShards(vid, "c", 0, bits(10, 11, 12, 13)) // all parity, over the 2/node cap
+	full.SetNoReceive()
+	topo.AddNode("n3", "dc1", "dc1:r1", 50)
+
+	moves := Plan(topo, Options{Ratio: ratio(10, 4)})
+	if len(moves) == 0 {
+		t.Fatal("expected moves off the over-cap nodes")
+	}
+	shed := false
+	for _, m := range moves {
+		if m.TargetNode == "n2" {
+			t.Fatalf("move targets the refusing node: %+v", m)
+		}
+		if m.SourceNode == "n2" {
+			shed = true
+		}
+	}
+	if !shed {
+		t.Error("the refusing node should still shed its over-cap shards as a source")
+	}
+}
+
+// TestNoReceiveDiskLeavesSiblingsEligible: an explicit-disk refusal marks only
+// that disk, not the node — the planner may still target the node's other
+// disks, but never the refused one.
+func TestNoReceiveDiskLeavesSiblingsEligible(t *testing.T) {
+	const vid = uint32(1)
+	topo := NewTopology()
+	src := topo.AddNode("n1", "dc1", "dc1:r1", 50)
+	src.AddShards(vid, "c", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))
+	n2 := topo.AddNode("n2", "dc1", "dc1:r1", 50)
+	// The refused disk scores better (empty); without the noReceive check the
+	// picker would choose it again.
+	n2.AddDisk(1, "hdd", 50, 0)
+	n2.AddDisk(2, "hdd", 50, 50)
+	n2.AddShards(vid, "c", 2, bits(10, 11, 12, 13))
+	n2.SetDiskNoReceive(1)
+	topo.AddNode("n3", "dc1", "dc1:r1", 50)
+
+	moves := Plan(topo, Options{Ratio: ratio(10, 4)})
+	targeted := false
+	for _, m := range moves {
+		if m.TargetNode != "n2" {
+			continue
+		}
+		targeted = true
+		if m.TargetDisk == 1 {
+			t.Fatalf("move targets the refused disk: %+v", m)
+		}
+	}
+	if !targeted {
+		t.Fatal("a node with one refused disk should still receive on its other disks")
+	}
+}

weed/storage/erasure_coding/ecbalancer/place.go

@@ -430,6 +430,9 @@ func rackTotalCap(rackKeys []string, held, room map[string]int, totalShards int)
 // its free slots, bounded by the per-disk durability cap (maxPerDisk shards of
 // one volume per disk; <= 0 disables it).
 func diskShardRoom(n *Node, d *disk, vk volKey, maxPerDisk int) int {
+	if d.noReceive {
+		return 0
+	}
 	room := d.freeSlots
 	if maxPerDisk > 0 {
 		held := 0
@@ -444,13 +447,16 @@ func diskShardRoom(n *Node, d *disk, vk volKey, maxPerDisk int) int {
 // nodeShardRoom returns how many more shards of the volume the node's eligible
 // disks can take, bounded by the node's free slots.
 func nodeShardRoom(n *Node, vk volKey, eligible func(*disk) bool, maxPerDisk int) int {
+	if n.noReceive {
+		return 0
+	}
 	room := 0
 	for _, d := range n.disks {
 		if eligible(d) {
 			room += diskShardRoom(n, d, vk, maxPerDisk)
 		}
 	}
-	return max(min(room, n.freeSlots), 0)
+	return max(min(room, n.receiveSlots()), 0)
 }
 
 // rackShardRoom returns how many more shards of the volume the rack can take.
@@ -532,7 +538,7 @@ func pickBestDiskEligible(node *Node, vk volKey, eligible func(*disk) bool, pref
 	bestPreferred := false
 	for _, diskID := range sortedDiskKeys(node.disks) {
 		d := node.disks[diskID]
-		if !eligible(d) || d.freeSlots <= 0 {
+		if !eligible(d) || d.freeSlots <= 0 || d.noReceive {
 			continue
 		}
 		existingShards := 0

weed/util/constants/filer.go

@@ -7,6 +7,10 @@ const (
 	ErrMsgOperationNotPermitted = "operation not permitted"
 	ErrMsgBadDigest             = "The Content-Md5 you specified did not match what we received."
 	ErrMsgEntryNameTooLong      = "entry name too long"
+	// Unsupported write-mode combinations; deterministic client errors.
+	ErrMsgAppendSaveInsideChunked = "append saved inside the entry to a chunked file is not supported"
+	ErrMsgAppendInlineMode        = "append to inline content with this storage mode is not supported"
+	ErrMsgOffsetWriteInline       = "offset write to inline small file is not supported yet"
 	// Suffix appended to a path, e.g. "/buckets/b/foo/bar is a file"
 	ErrMsgIsAFile = " is a file"
 	// Prefix + suffix, e.g. "existing /buckets/b/foo is a directory"

weed/worker/tasks/ec_balance/ec_balance_task.go

@@ -116,17 +116,29 @@ func (t *ECBalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParam
 		return nil
 	}
 
-	err := mover.MoveEcShards(ctx, volume_move.EcShardMove{
+	move := volume_move.EcShardMove{
 		VolumeId:   needle.VolumeId(params.VolumeId),
 		Collection: params.Collection,
 		ShardIds:   shardIds,
 		Source:     sourceAddr,
 		Target:     targetAddr,
 		TargetDisk: target.DiskId,
-	}, volume_move.EcMoveOptions{
+	}
+	opts := volume_move.EcMoveOptions{
 		IoBytePerSecond: ecParams.GetIoBytePerSecond(),
 		Progress:        t.reportProgress,
-	})
+	}
+	err := mover.MoveEcShards(ctx, move, opts)
+	// A "no space left" refusal to a pinned disk convicts only that disk;
+	// retry once letting the server pick any disk on the same node. The
+	// planner's node choice stands; only the intra-node disk defers to the
+	// server's live view. When the node is truly full, the retry fails with
+	// the same error.
+	if err != nil && target.DiskId > 0 && volume_move.IsNoSpaceLeftError(err) {
+		glog.Infof("EC balance: %s disk %d has no space left; retrying with server-picked disk", target.Node, target.DiskId)
+		move.TargetDisk = 0
+		err = mover.MoveEcShards(ctx, move, opts)
+	}
 	if err != nil {
 		return err
 	}

weed/worker/tasks/ec_balance/ec_balance_task_test.go

@@ -6,6 +6,8 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func TestECBalanceTaskRejectsOutOfRangeShardIds(t *testing.T) {
@@ -25,3 +27,32 @@ func TestECBalanceTaskRejectsOutOfRangeShardIds(t *testing.T) {
 		t.Fatalf("Execute: expected out-of-range rejection, got: %v", err)
 	}
 }
+
+// A refusal against the pinned disk convicts only that disk, not the node:
+// the task retries once letting the server pick a disk, and the move completes.
+func TestECBalanceTaskRetriesServerPickOnPinnedDiskRefusal(t *testing.T) {
+	source := startFakeEcVolumeServer(t, 7, 5)
+	target := startFakeEcVolumeServer(t, 7)
+	target.refuseCopyOnDisk = 2
+
+	task := NewECBalanceTask("t2", 7, "c1",
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	params := &worker_pb.TaskParams{
+		VolumeId: 7,
+		Sources:  []*worker_pb.TaskSource{{Node: source.address(), ShardIds: []uint32{5}}},
+		Targets:  []*worker_pb.TaskTarget{{Node: target.address(), DiskId: 2, ShardIds: []uint32{5}}},
+	}
+
+	if err := task.Execute(context.Background(), params); err != nil {
+		t.Fatalf("Execute: %v", err)
+	}
+	if got := target.copyDiskIds; len(got) != 2 || got[0] != 2 || got[1] != 0 {
+		t.Fatalf("expected copies to disk 2 then server-picked 0, got %v", got)
+	}
+	if source.has(7, 5) {
+		t.Fatalf("source still holds shard 5")
+	}
+	if !target.has(7, 5) {
+		t.Fatalf("target never received shard 5")
+	}
+}

weed/worker/tasks/ec_balance/fake_ec_volume_server_test.go

@@ -32,6 +32,12 @@ type fakeEcVolumeServer struct {
 	// the request even when it was a no-op against the inventory.
 	deleted   []string
 	unmounted []string
+	// refuseCopyOnDisk, when nonzero, makes VolumeEcShardsCopy fail requests
+	// pinning that disk id with "no space left" — a full disk; a request with
+	// DiskId 0 (server picks) still succeeds.
+	refuseCopyOnDisk uint32
+	// copyDiskIds records the DiskId each copy request carried.
+	copyDiskIds []uint32
 	// grpcAddr is host:port; the code under test dials by address string.
 	grpcAddr string
 	stop     func()
@@ -46,11 +52,31 @@ func (f *fakeEcVolumeServer) VolumeEcShardsInfo(ctx context.Context, req *volume
 			VolumeId:   req.VolumeId,
 			ShardId:    sid,
 			Collection: f.collection,
+			Size:       1,
 		})
 	}
 	return resp, nil
 }
 
+// VolumeEcShardsCopy records the request's DiskId and adds the shards to this
+// server's inventory, unless the pinned disk is the configured refusing disk.
+func (f *fakeEcVolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.copyDiskIds = append(f.copyDiskIds, req.DiskId)
+	if f.refuseCopyOnDisk != 0 && req.DiskId == f.refuseCopyOnDisk {
+		return nil, fmt.Errorf("disk %d: no space left", req.DiskId)
+	}
+	for _, sid := range req.ShardIds {
+		f.shards[req.VolumeId][sid] = true
+	}
+	return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil
+}
+
+func (f *fakeEcVolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) {
+	return &volume_server_pb.VolumeEcShardsMountResponse{}, nil
+}
+
 func (f *fakeEcVolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()