refactor: remove withoutCancelIfAlive helper, inline ctx.Err() checks directly

The helper didn't reduce line count — just inline the straightforward
ctx.Err() check at each call site.

.github/workflows/rust-volume-server-tests.yml

@@ -93,7 +93,7 @@ jobs:
       - name: Build Go weed binary
         run: |
           cd weed
-          go build -tags 5BytesOffset -o weed .
+          go build -o weed .
           chmod +x weed
           ./weed version
 
@@ -169,7 +169,7 @@ jobs:
       - name: Build Go weed binary
         run: |
           cd weed
-          go build -tags 5BytesOffset -o weed .
+          go build -o weed .
           chmod +x weed
           ./weed version
 

docker/compose/s3_static_users_example.json

@@ -1,53 +0,0 @@
-{
-  "identities": [
-    {
-      "name": "admin",
-      "credentials": [
-        {
-          "accessKey": "AKIAIOSFODNN7EXAMPLE",
-          "secretKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
-        }
-      ],
-      "actions": [
-        "Admin",
-        "Read",
-        "List",
-        "Tagging",
-        "Write"
-      ]
-    },
-    {
-      "name": "steward",
-      "credentials": [
-        {
-          "accessKey": "steward-key",
-          "secretKey": "steward-secret"
-        }
-      ],
-      "actions": [
-        "Read",
-        "List",
-        "Write"
-      ]
-    },
-    {
-      "name": "le001",
-      "credentials": [
-        {
-          "accessKey": "le001-key",
-          "secretKey": "le001-secret"
-        }
-      ],
-      "actions": [
-        "Read",
-        "List"
-      ]
-    },
-    {
-      "name": "anonymous",
-      "actions": [
-        "Read"
-      ]
-    }
-  ]
-}

seaweed-volume/Cargo.toml

@@ -20,7 +20,7 @@ default = ["5bytes"]
 [dependencies]
 # Async runtime
 tokio = { version = "1", features = ["full"] }
-tokio-stream = { version = "0.1", features = ["net"] }
+tokio-stream = "0.1"
 tokio-io-timeout = "1"
 
 # gRPC + protobuf

seaweed-volume/src/main.rs

@@ -600,35 +600,35 @@ async fn run(
         })
     };
 
-    // Bind the gRPC listener before spawning to propagate bind errors at startup.
-    let grpc_listener = tokio::net::TcpListener::bind(&grpc_addr)
-        .await
-        .unwrap_or_else(|e| panic!("Failed to bind gRPC to {}: {}", grpc_addr, e));
-    let grpc_local_addr = grpc_listener
-        .local_addr()
-        .unwrap_or_else(|e| panic!("Failed to get gRPC local addr: {}", e));
-
     let grpc_handle = {
         let grpc_state = state.clone();
+        let grpc_addr = grpc_addr.clone();
         let grpc_tls_acceptor = grpc_tls_acceptor.clone();
         let mut shutdown_rx = shutdown_tx.subscribe();
-        let shutdown_tx_grpc = shutdown_tx.clone();
         tokio::spawn(async move {
+            let addr = tokio::net::lookup_host(&grpc_addr)
+                .await
+                .expect("Failed to resolve gRPC address")
+                .next()
+                .expect("No addresses found for gRPC bind address");
             let grpc_service = VolumeGrpcService {
                 state: grpc_state.clone(),
             };
-            let reflection_v1 = tonic_reflection::server::Builder::configure()
-                .register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
-                .build_v1()
-                .expect("Failed to build gRPC reflection v1 service");
-            let reflection_v1alpha = tonic_reflection::server::Builder::configure()
-                .register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
-                .build_v1alpha()
-                .expect("Failed to build gRPC reflection v1alpha service");
-            let result = if let Some(tls_acceptor) = grpc_tls_acceptor {
-                let incoming = grpc_tls_incoming(grpc_listener, tls_acceptor);
-                info!("gRPC server listening on {} (TLS enabled)", grpc_local_addr);
-                build_grpc_server_builder()
+            if let Some(tls_acceptor) = grpc_tls_acceptor {
+                let listener = tokio::net::TcpListener::bind(&grpc_addr)
+                    .await
+                    .unwrap_or_else(|e| panic!("Failed to bind gRPC to {}: {}", grpc_addr, e));
+                let incoming = grpc_tls_incoming(listener, tls_acceptor);
+                let reflection_v1 = tonic_reflection::server::Builder::configure()
+                    .register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
+                    .build_v1()
+                    .expect("Failed to build gRPC reflection v1 service");
+                let reflection_v1alpha = tonic_reflection::server::Builder::configure()
+                    .register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
+                    .build_v1alpha()
+                    .expect("Failed to build gRPC reflection v1alpha service");
+                info!("gRPC server listening on {} (TLS enabled)", addr);
+                if let Err(e) = build_grpc_server_builder()
                     .layer(GrpcRequestIdLayer)
                     .add_service(reflection_v1)
                     .add_service(reflection_v1alpha)
@@ -637,25 +637,32 @@ async fn run(
                         let _ = shutdown_rx.recv().await;
                     })
                     .await
+                {
+                    error!("gRPC server error: {}", e);
+                }
             } else {
-                let incoming =
-                    tokio_stream::wrappers::TcpListenerStream::new(grpc_listener);
-                info!("gRPC server listening on {}", grpc_local_addr);
-                build_grpc_server_builder()
+                let reflection_v1 = tonic_reflection::server::Builder::configure()
+                    .register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
+                    .build_v1()
+                    .expect("Failed to build gRPC reflection v1 service");
+                let reflection_v1alpha = tonic_reflection::server::Builder::configure()
+                    .register_encoded_file_descriptor_set(seaweed_volume::pb::FILE_DESCRIPTOR_SET)
+                    .build_v1alpha()
+                    .expect("Failed to build gRPC reflection v1alpha service");
+                info!("gRPC server listening on {}", addr);
+                if let Err(e) = build_grpc_server_builder()
                     .layer(GrpcRequestIdLayer)
                     .add_service(reflection_v1)
                     .add_service(reflection_v1alpha)
                     .add_service(build_volume_grpc_service(grpc_service))
-                    .serve_with_incoming_shutdown(incoming, async move {
+                    .serve_with_shutdown(addr, async move {
                         let _ = shutdown_rx.recv().await;
                     })
                     .await
-            };
-            if let Err(ref e) = result {
-                error!("gRPC server error: {}", e);
-                let _ = shutdown_tx_grpc.send(());
+                {
+                    error!("gRPC server error: {}", e);
+                }
             }
-            result
         })
     };
 
@@ -764,40 +771,9 @@ async fn run(
         }))
     };
 
-    // Wait for servers. Use select! with &mut so the losing handle is not
-    // dropped, then await it explicitly afterward.
-    let mut server_err: Option<String> = None;
-    let mut http_handle = http_handle;
-    let mut grpc_handle = grpc_handle;
-    let grpc_finished_first = tokio::select! {
-        _ = &mut http_handle => false,
-        _ = &mut grpc_handle => true,
-    };
-    // Inspect the gRPC result (already resolved if it finished first,
-    // otherwise await it now).
-    let grpc_result = if grpc_finished_first {
-        grpc_handle.await
-    } else {
-        // HTTP finished first; gRPC is still running. Await it.
-        grpc_handle.await
-    };
-    match grpc_result {
-        Ok(Ok(())) => {}
-        Ok(Err(e)) => {
-            let msg = format!("gRPC server exited with error: {}", e);
-            error!("{}", msg);
-            server_err = Some(msg);
-            // serve error already sent shutdown inside the task
-        }
-        Err(e) => {
-            let msg = format!("gRPC task panicked: {}", e);
-            error!("{}", msg);
-            server_err = Some(msg);
-            let _ = shutdown_tx.send(());
-        }
-    }
-    // Ensure the HTTP handle completes too.
+    // Wait for all servers
     let _ = http_handle.await;
+    let _ = grpc_handle.await;
     if let Some(h) = public_handle {
         let _ = h.await;
     }
@@ -822,10 +798,6 @@ async fn run(
         cpu_profile.finish().map_err(std::io::Error::other)?;
     }
 
-    if let Some(err_msg) = server_err {
-        return Err(std::io::Error::other(err_msg).into());
-    }
-
     info!("Volume server stopped.");
     Ok(())
 }

seaweed-volume/src/server/grpc_server.rs

@@ -1867,7 +1867,6 @@ impl VolumeServer for VolumeGrpcService {
         {
             let needle_header = resp.needle_header;
             let mut needle_body = resp.needle_body;
-            let resp_version = resp.version;
 
             if needle_header.is_empty() {
                 continue;
@@ -1892,36 +1891,8 @@ impl VolumeServer for VolumeGrpcService {
             // Parse needle from header + body
             let mut n = Needle::default();
             n.read_header(&needle_header);
-
-            if n.size.0 < 0 {
-                return Err(Status::invalid_argument(format!(
-                    "unexpected negative needle size {} for needle {}",
-                    n.size.0, n.id.0
-                )));
-            } else if n.size.0 > 0 {
-                // Normal needle: parse the body fields (DataSize, Data, flags, etc.)
-                n.read_body_v2(&needle_body)
-                    .map_err(|e| Status::internal(format!("parse needle body: {}", e)))?;
-            } else {
-                // Delete tombstone (size == 0): body is checksum + timestamp
-                // (V3) or checksum only (V2) + padding. Validate minimum
-                // footer length for the protocol version.
-                use crate::storage::types::{
-                    NEEDLE_CHECKSUM_SIZE, TIMESTAMP_SIZE, VERSION_3, Version,
-                };
-                let version = Version(resp_version as u8);
-                let min_footer = if version >= VERSION_3 {
-                    NEEDLE_CHECKSUM_SIZE + TIMESTAMP_SIZE
-                } else {
-                    NEEDLE_CHECKSUM_SIZE
-                };
-                if needle_body.len() < min_footer {
-                    return Err(Status::invalid_argument(format!(
-                        "tombstone needle {} body too short: got {} bytes, need >= {} for version {}",
-                        n.id.0, needle_body.len(), min_footer, resp_version
-                    )));
-                }
-            }
+            n.read_body_v2(&needle_body)
+                .map_err(|e| Status::internal(format!("parse needle body: {}", e)))?;
 
             // Write needle to local volume
             let mut store = state.store.write().unwrap();
@@ -4076,12 +4047,11 @@ fn find_last_append_at_ns(idx_path: &str, dat_path: &str, version: u32) -> Optio
     let mut header = [0u8; 16];
     dat_file.read_exact(&mut header).ok()?;
     let needle_size = i32::from_be_bytes([header[12], header[13], header[14], header[15]]);
-    if needle_size < 0 {
+    if needle_size <= 0 {
         return None;
     }
 
     // Seek to tail: offset + 16 (header) + size -> checksum (4) + timestamp (8)
-    // For delete needles (size == 0), the tail is right after the header.
     let tail_offset = actual_offset as u64 + 16 + needle_size as u64;
     dat_file.seek(SeekFrom::Start(tail_offset)).ok()?;
 

test/volume_server/framework/cluster_mixed.go

@@ -1,368 +0,0 @@
-package framework
-
-import (
-	"fmt"
-	"net"
-	"os"
-	"os/exec"
-	"path/filepath"
-	"strconv"
-	"sync"
-	"testing"
-
-	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
-)
-
-// MixedVolumeCluster wraps a Go master + a mix of Go and Rust volume servers
-// for integration testing. Go servers occupy indices [0, goCount) and Rust
-// servers occupy indices [goCount, goCount+rustCount).
-type MixedVolumeCluster struct {
-	testingTB testing.TB
-	profile   matrix.Profile
-
-	weedBinary       string // Go weed binary (master + Go volume servers)
-	rustVolumeBinary string // Rust volume binary
-
-	baseDir   string
-	configDir string
-	logsDir   string
-	keepLogs  bool
-
-	masterPort     int
-	masterGrpcPort int
-
-	volumePorts     []int
-	volumeGrpcPorts []int
-	volumePubPorts  []int
-	isRust          []bool // which servers are Rust
-
-	masterCmd  *exec.Cmd
-	volumeCmds []*exec.Cmd
-
-	cleanupOnce sync.Once
-}
-
-// StartMixedVolumeCluster starts a cluster with 1 Go master, goCount Go volume
-// servers, and rustCount Rust volume servers. Go servers come first in the index.
-func StartMixedVolumeCluster(t testing.TB, profile matrix.Profile, goCount, rustCount int) *MixedVolumeCluster {
-	t.Helper()
-
-	if goCount < 0 || rustCount < 0 {
-		t.Fatalf("goCount and rustCount must be non-negative, got go=%d rust=%d", goCount, rustCount)
-	}
-	total := goCount + rustCount
-	if total < 2 {
-		t.Fatalf("need at least 2 volume servers, got %d", total)
-	}
-
-	weedBinary, err := FindOrBuildWeedBinary()
-	if err != nil {
-		t.Fatalf("resolve weed binary: %v", err)
-	}
-
-	// Only build the Rust binary when Rust servers are requested.
-	var rustBinary string
-	if rustCount > 0 {
-		rustBinary, err = FindOrBuildRustBinary()
-		if err != nil {
-			t.Skipf("skipping mixed cluster test: rust binary unavailable: %v", err)
-		}
-	}
-
-	baseDir, keepLogs, err := newWorkDir()
-	if err != nil {
-		t.Fatalf("create temp test directory: %v", err)
-	}
-
-	configDir := filepath.Join(baseDir, "config")
-	logsDir := filepath.Join(baseDir, "logs")
-	masterDataDir := filepath.Join(baseDir, "master")
-
-	dirs := []string{configDir, logsDir, masterDataDir}
-	for i := 0; i < total; i++ {
-		dirs = append(dirs, filepath.Join(baseDir, fmt.Sprintf("volume%d", i)))
-	}
-	for _, dir := range dirs {
-		if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
-			t.Fatalf("create %s: %v", dir, mkErr)
-		}
-	}
-
-	if err = writeSecurityConfig(configDir, profile); err != nil {
-		t.Fatalf("write security config: %v", err)
-	}
-
-	masterPort, masterGrpcPort, err := allocateMasterPortPair()
-	if err != nil {
-		t.Fatalf("allocate master port pair: %v", err)
-	}
-
-	// 2 ports per server (admin, grpc); add 1 more when public port is split out.
-	portsPerServer := 2
-	if profile.SplitPublicPort {
-		portsPerServer = 3
-	}
-	ports, err := allocatePorts(total * portsPerServer)
-	if err != nil {
-		t.Fatalf("allocate volume ports: %v", err)
-	}
-
-	isRust := make([]bool, total)
-	for i := goCount; i < total; i++ {
-		isRust[i] = true
-	}
-
-	c := &MixedVolumeCluster{
-		testingTB:        t,
-		profile:          profile,
-		weedBinary:       weedBinary,
-		rustVolumeBinary: rustBinary,
-		baseDir:          baseDir,
-		configDir:        configDir,
-		logsDir:          logsDir,
-		keepLogs:         keepLogs,
-		masterPort:       masterPort,
-		masterGrpcPort:   masterGrpcPort,
-		volumePorts:      make([]int, total),
-		volumeGrpcPorts:  make([]int, total),
-		volumePubPorts:   make([]int, total),
-		isRust:           isRust,
-		volumeCmds:       make([]*exec.Cmd, total),
-	}
-
-	for i := 0; i < total; i++ {
-		baseIdx := i * portsPerServer
-		c.volumePorts[i] = ports[baseIdx]
-		c.volumeGrpcPorts[i] = ports[baseIdx+1]
-		if profile.SplitPublicPort {
-			c.volumePubPorts[i] = ports[baseIdx+2]
-		} else {
-			c.volumePubPorts[i] = c.volumePorts[i] // reuse admin port
-		}
-	}
-
-	// Start master
-	if err = c.startMaster(masterDataDir); err != nil {
-		c.Stop()
-		t.Fatalf("start master: %v", err)
-	}
-	helper := &Cluster{logsDir: logsDir}
-	if err = helper.waitForHTTP(c.MasterURL() + "/dir/status"); err != nil {
-		masterLog := helper.tailLog("master.log")
-		c.Stop()
-		t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
-	}
-
-	// Start volume servers
-	for i := 0; i < total; i++ {
-		volumeDataDir := filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
-		if isRust[i] {
-			err = c.startRustVolume(i, volumeDataDir)
-		} else {
-			err = c.startGoVolume(i, volumeDataDir)
-		}
-		if err != nil {
-			logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
-			c.Stop()
-			t.Fatalf("start volume server %d (rust=%v): %v\nlog tail:\n%s", i, isRust[i], err, logTail)
-		}
-
-		// Rust uses /healthz, Go uses /status
-		healthURL := c.VolumeAdminURL(i) + "/status"
-		if isRust[i] {
-			healthURL = c.VolumeAdminURL(i) + "/healthz"
-		}
-		if err = helper.waitForHTTP(healthURL); err != nil {
-			logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
-			c.Stop()
-			t.Fatalf("wait for volume server %d readiness: %v\nlog tail:\n%s", i, err, logTail)
-		}
-		if err = helper.waitForTCP(c.VolumeGRPCAddress(i)); err != nil {
-			logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
-			c.Stop()
-			t.Fatalf("wait for volume server %d grpc readiness: %v\nlog tail:\n%s", i, err, logTail)
-		}
-	}
-
-	t.Cleanup(func() {
-		c.Stop()
-	})
-
-	return c
-}
-
-func (c *MixedVolumeCluster) Stop() {
-	if c == nil {
-		return
-	}
-	c.cleanupOnce.Do(func() {
-		for i := len(c.volumeCmds) - 1; i >= 0; i-- {
-			stopProcess(c.volumeCmds[i])
-		}
-		stopProcess(c.masterCmd)
-		if !c.keepLogs && !c.testingTB.Failed() {
-			_ = os.RemoveAll(c.baseDir)
-		} else if c.baseDir != "" {
-			c.testingTB.Logf("mixed volume server integration logs kept at %s", c.baseDir)
-		}
-	})
-}
-
-func (c *MixedVolumeCluster) startMaster(dataDir string) error {
-	logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
-	if err != nil {
-		return err
-	}
-
-	args := []string{
-		"-config_dir=" + c.configDir,
-		"master",
-		"-ip=127.0.0.1",
-		"-port=" + strconv.Itoa(c.masterPort),
-		"-port.grpc=" + strconv.Itoa(c.masterGrpcPort),
-		"-mdir=" + dataDir,
-		"-peers=none",
-		"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
-		"-defaultReplication=000",
-	}
-
-	c.masterCmd = exec.Command(c.weedBinary, args...)
-	c.masterCmd.Dir = c.baseDir
-	c.masterCmd.Stdout = logFile
-	c.masterCmd.Stderr = logFile
-	if err = c.masterCmd.Start(); err != nil {
-		logFile.Close()
-		return err
-	}
-	logFile.Close() // child inherited the fd
-	return nil
-}
-
-func (c *MixedVolumeCluster) startGoVolume(index int, dataDir string) error {
-	logName := fmt.Sprintf("volume%d.log", index)
-	logFile, err := os.Create(filepath.Join(c.logsDir, logName))
-	if err != nil {
-		return err
-	}
-
-	args := []string{
-		"-config_dir=" + c.configDir,
-		"volume",
-		"-ip=127.0.0.1",
-		"-port=" + strconv.Itoa(c.volumePorts[index]),
-		"-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]),
-		"-port.public=" + strconv.Itoa(c.volumePubPorts[index]),
-		"-dir=" + dataDir,
-		"-max=16",
-		"-master=127.0.0.1:" + strconv.Itoa(c.masterPort),
-		"-readMode=" + c.profile.ReadMode,
-		"-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB),
-		"-concurrentDownloadLimitMB=" + strconv.Itoa(c.profile.ConcurrentDownloadLimitMB),
-	}
-	if c.profile.InflightUploadTimeout > 0 {
-		args = append(args, "-inflightUploadDataTimeout="+c.profile.InflightUploadTimeout.String())
-	}
-	if c.profile.InflightDownloadTimeout > 0 {
-		args = append(args, "-inflightDownloadDataTimeout="+c.profile.InflightDownloadTimeout.String())
-	}
-
-	cmd := exec.Command(c.weedBinary, args...)
-	cmd.Dir = c.baseDir
-	cmd.Stdout = logFile
-	cmd.Stderr = logFile
-	if err = cmd.Start(); err != nil {
-		logFile.Close()
-		return err
-	}
-	logFile.Close() // child inherited the fd
-	c.volumeCmds[index] = cmd
-	return nil
-}
-
-func (c *MixedVolumeCluster) startRustVolume(index int, dataDir string) error {
-	logName := fmt.Sprintf("volume%d.log", index)
-	logFile, err := os.Create(filepath.Join(c.logsDir, logName))
-	if err != nil {
-		return err
-	}
-
-	args := rustVolumeArgs(
-		c.profile,
-		c.configDir,
-		c.masterPort,
-		c.volumePorts[index],
-		c.volumeGrpcPorts[index],
-		c.volumePubPorts[index],
-		dataDir,
-	)
-
-	cmd := exec.Command(c.rustVolumeBinary, args...)
-	cmd.Dir = c.baseDir
-	cmd.Stdout = logFile
-	cmd.Stderr = logFile
-	if err = cmd.Start(); err != nil {
-		logFile.Close()
-		return err
-	}
-	logFile.Close() // child inherited the fd
-	c.volumeCmds[index] = cmd
-	return nil
-}
-
-// --- accessor methods (mirror MultiVolumeCluster) ---
-
-func (c *MixedVolumeCluster) MasterAddress() string {
-	return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.masterPort))
-}
-
-func (c *MixedVolumeCluster) MasterURL() string {
-	return "http://" + c.MasterAddress()
-}
-
-func (c *MixedVolumeCluster) VolumeAdminAddress(index int) string {
-	if index < 0 || index >= len(c.volumePorts) {
-		return ""
-	}
-	return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePorts[index]))
-}
-
-func (c *MixedVolumeCluster) VolumePublicAddress(index int) string {
-	if index < 0 || index >= len(c.volumePubPorts) {
-		return ""
-	}
-	return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPorts[index]))
-}
-
-func (c *MixedVolumeCluster) VolumeGRPCAddress(index int) string {
-	if index < 0 || index >= len(c.volumeGrpcPorts) {
-		return ""
-	}
-	return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPorts[index]))
-}
-
-func (c *MixedVolumeCluster) VolumeAdminURL(index int) string {
-	return "http://" + c.VolumeAdminAddress(index)
-}
-
-func (c *MixedVolumeCluster) VolumePublicURL(index int) string {
-	return "http://" + c.VolumePublicAddress(index)
-}
-
-func (c *MixedVolumeCluster) BaseDir() string {
-	return c.baseDir
-}
-
-// VolumeServerAddress returns SeaweedFS server address format: ip:httpPort.grpcPort
-func (c *MixedVolumeCluster) VolumeServerAddress(index int) string {
-	if index < 0 || index >= len(c.volumePorts) {
-		return ""
-	}
-	return fmt.Sprintf("%s.%d", c.VolumeAdminAddress(index), c.volumeGrpcPorts[index])
-}
-
-func (c *MixedVolumeCluster) IsRust(index int) bool {
-	if index < 0 || index >= len(c.isRust) {
-		return false
-	}
-	return c.isRust[index]
-}

test/volume_server/grpc/mixed_balance_test.go

@@ -1,567 +0,0 @@
-package volume_server_grpc_test
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"net/http"
-	"testing"
-	"time"
-
-	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
-	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
-	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
-)
-
-// deleteAndWaitForTombstone issues an HTTP DELETE for the given fid on
-// volumeURL, asserts a successful response, then polls with GET until
-// the file returns 404 (tombstone visible) or the timeout elapses.
-func deleteAndWaitForTombstone(t *testing.T, httpClient *http.Client, volumeURL, fid string) {
-	t.Helper()
-	req, err := http.NewRequest(http.MethodDelete, fmt.Sprintf("%s/%s", volumeURL, fid), nil)
-	if err != nil {
-		t.Fatalf("build delete request for %s: %v", fid, err)
-	}
-	resp := framework.DoRequest(t, httpClient, req)
-	framework.ReadAllAndClose(t, resp)
-	if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusOK {
-		t.Fatalf("delete %s: expected 200 or 202, got %d", fid, resp.StatusCode)
-	}
-	// Poll until GET returns 404 (tombstone flushed to disk).
-	deadline := time.Now().Add(5 * time.Second)
-	for time.Now().Before(deadline) {
-		getResp := framework.ReadBytes(t, httpClient, volumeURL, fid)
-		status := getResp.StatusCode
-		framework.ReadAllAndClose(t, getResp)
-		if status == http.StatusNotFound {
-			return
-		}
-		time.Sleep(50 * time.Millisecond)
-	}
-	t.Fatalf("delete %s: tombstone not visible after 5s", fid)
-}
-
-// TestMixedBalanceCopyGoToRust verifies that VolumeCopy works from a Go volume
-// server to a Rust volume server. This is the core operation behind volume
-// balancing in a mixed Go+Rust cluster.
-func TestMixedBalanceCopyGoToRust(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test in short mode")
-	}
-
-	cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
-
-	// server 0 = Go, server 1 = Rust
-	conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
-	defer conn0.Close()
-	conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
-	defer conn1.Close()
-
-	httpClient := framework.NewHTTPClient()
-
-	const volumeID = uint32(50)
-
-	// Allocate volume on Go server and upload test data
-	framework.AllocateVolume(t, goClient, volumeID, "")
-
-	testFiles := []struct {
-		key    uint64
-		cookie uint32
-		data   []byte
-	}{
-		{1, 0xAABBCCDD, []byte("hello from Go server")},
-		{2, 0x11223344, []byte("second file for balance test")},
-		{3, 0xDEADBEEF, make([]byte, 4096)}, // larger file
-	}
-
-	for _, f := range testFiles {
-		fid := framework.NewFileID(volumeID, f.key, f.cookie)
-		resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid, f.data)
-		body := framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusCreated {
-			t.Fatalf("upload %s: expected 201, got %d: %s", fid, resp.StatusCode, body)
-		}
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
-	// Read source volume status before copy
-	sourceStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus on Go server: %v", err)
-	}
-	t.Logf("Source: dat=%d idx=%d files=%d version=%d",
-		sourceStatus.GetDatFileSize(), sourceStatus.GetIdxFileSize(),
-		sourceStatus.GetFileCount(), sourceStatus.GetVersion())
-
-	// Copy volume from Go (server 0) to Rust (server 1)
-	copyStream, err := rustClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
-		VolumeId:       volumeID,
-		SourceDataNode: cluster.VolumeServerAddress(0),
-	})
-	if err != nil {
-		t.Fatalf("VolumeCopy call failed: %v", err)
-	}
-
-	var lastAppendAtNs uint64
-	for {
-		resp, recvErr := copyStream.Recv()
-		if recvErr != nil {
-			if recvErr != io.EOF {
-				t.Fatalf("VolumeCopy recv error: %v", recvErr)
-			}
-			break
-		}
-		if resp.GetLastAppendAtNs() != 0 {
-			lastAppendAtNs = resp.GetLastAppendAtNs()
-		}
-	}
-	t.Logf("VolumeCopy completed, lastAppendAtNs=%d", lastAppendAtNs)
-
-	// Verify: read volume status from Rust server
-	targetStatus, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus on Rust server: %v", err)
-	}
-	t.Logf("Target: dat=%d idx=%d files=%d version=%d",
-		targetStatus.GetDatFileSize(), targetStatus.GetIdxFileSize(),
-		targetStatus.GetFileCount(), targetStatus.GetVersion())
-
-	if sourceStatus.GetDatFileSize() != targetStatus.GetDatFileSize() {
-		t.Fatalf("dat file size mismatch: source=%d target=%d",
-			sourceStatus.GetDatFileSize(), targetStatus.GetDatFileSize())
-	}
-	if sourceStatus.GetIdxFileSize() != targetStatus.GetIdxFileSize() {
-		t.Fatalf("idx file size mismatch: source=%d target=%d",
-			sourceStatus.GetIdxFileSize(), targetStatus.GetIdxFileSize())
-	}
-	if sourceStatus.GetFileCount() != targetStatus.GetFileCount() {
-		t.Fatalf("file count mismatch: source=%d target=%d",
-			sourceStatus.GetFileCount(), targetStatus.GetFileCount())
-	}
-
-	// Verify data can be read from Rust server
-	for _, f := range testFiles {
-		fid := framework.NewFileID(volumeID, f.key, f.cookie)
-		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
-		body := framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusOK {
-			t.Fatalf("read %s from Rust server: expected 200, got %d", fid, resp.StatusCode)
-		}
-		if !bytes.Equal(body, f.data) {
-			t.Fatalf("read %s from Rust server: content mismatch (got %d bytes, want %d)", fid, len(body), len(f.data))
-		}
-	}
-}
-
-// TestMixedBalanceCopyRustToGo verifies that VolumeCopy works from a Rust
-// volume server to a Go volume server.
-func TestMixedBalanceCopyRustToGo(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test in short mode")
-	}
-
-	cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
-
-	conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
-	defer conn0.Close()
-	conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
-	defer conn1.Close()
-
-	httpClient := framework.NewHTTPClient()
-
-	const volumeID = uint32(51)
-
-	// Allocate volume on Rust server and upload test data
-	framework.AllocateVolume(t, rustClient, volumeID, "")
-
-	testFiles := []struct {
-		key    uint64
-		cookie uint32
-		data   []byte
-	}{
-		{1, 0xAABBCCDD, []byte("hello from Rust server")},
-		{2, 0x11223344, []byte("second file for reverse balance")},
-	}
-
-	for _, f := range testFiles {
-		fid := framework.NewFileID(volumeID, f.key, f.cookie)
-		resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid, f.data)
-		body := framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusCreated {
-			t.Fatalf("upload %s: expected 201, got %d: %s", fid, resp.StatusCode, body)
-		}
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
-	sourceStatus, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus on Rust server: %v", err)
-	}
-
-	// Copy volume from Rust (server 1) to Go (server 0)
-	copyStream, err := goClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
-		VolumeId:       volumeID,
-		SourceDataNode: cluster.VolumeServerAddress(1),
-	})
-	if err != nil {
-		t.Fatalf("VolumeCopy call failed: %v", err)
-	}
-	for {
-		_, recvErr := copyStream.Recv()
-		if recvErr != nil {
-			if recvErr != io.EOF {
-				t.Fatalf("VolumeCopy recv error: %v", recvErr)
-			}
-			break
-		}
-	}
-
-	targetStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus on Go server: %v", err)
-	}
-
-	if sourceStatus.GetDatFileSize() != targetStatus.GetDatFileSize() {
-		t.Fatalf("dat file size mismatch: source=%d target=%d",
-			sourceStatus.GetDatFileSize(), targetStatus.GetDatFileSize())
-	}
-	if sourceStatus.GetIdxFileSize() != targetStatus.GetIdxFileSize() {
-		t.Fatalf("idx file size mismatch: source=%d target=%d",
-			sourceStatus.GetIdxFileSize(), targetStatus.GetIdxFileSize())
-	}
-	if sourceStatus.GetFileCount() != targetStatus.GetFileCount() {
-		t.Fatalf("file count mismatch: source=%d target=%d",
-			sourceStatus.GetFileCount(), targetStatus.GetFileCount())
-	}
-
-	// Verify data can be read from Go server
-	for _, f := range testFiles {
-		fid := framework.NewFileID(volumeID, f.key, f.cookie)
-		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid)
-		body := framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusOK {
-			t.Fatalf("read %s from Go server: expected 200, got %d", fid, resp.StatusCode)
-		}
-		if !bytes.Equal(body, f.data) {
-			t.Fatalf("read %s from Go server: content mismatch (got %d bytes, want %d)", fid, len(body), len(f.data))
-		}
-	}
-}
-
-// TestMixedBalanceCopyWithDeletes verifies that VolumeCopy correctly handles
-// volumes that have both active and deleted needles.
-func TestMixedBalanceCopyWithDeletes(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test in short mode")
-	}
-
-	cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
-
-	conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
-	defer conn0.Close()
-	conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
-	defer conn1.Close()
-
-	httpClient := framework.NewHTTPClient()
-
-	const volumeID = uint32(52)
-	framework.AllocateVolume(t, goClient, volumeID, "")
-
-	// Upload files
-	for i := uint64(1); i <= 5; i++ {
-		fid := framework.NewFileID(volumeID, i, 0x12340000+uint32(i))
-		resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid, []byte(fmt.Sprintf("file-%d", i)))
-		framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusCreated {
-			t.Fatalf("upload file %d: expected 201, got %d", i, resp.StatusCode)
-		}
-	}
-
-	// Delete some files and wait for tombstones to be visible
-	for _, key := range []uint64{2, 4} {
-		fid := framework.NewFileID(volumeID, key, 0x12340000+uint32(key))
-		deleteAndWaitForTombstone(t, httpClient, cluster.VolumeAdminURL(0), fid)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
-	sourceStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus: %v", err)
-	}
-	t.Logf("Source after deletes: dat=%d idx=%d files=%d",
-		sourceStatus.GetDatFileSize(), sourceStatus.GetIdxFileSize(),
-		sourceStatus.GetFileCount())
-
-	// Copy volume from Go to Rust
-	copyStream, err := rustClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
-		VolumeId:       volumeID,
-		SourceDataNode: cluster.VolumeServerAddress(0),
-	})
-	if err != nil {
-		t.Fatalf("VolumeCopy: %v", err)
-	}
-	var lastAppendAtNs uint64
-	for {
-		resp, recvErr := copyStream.Recv()
-		if recvErr != nil {
-			if recvErr != io.EOF {
-				t.Fatalf("VolumeCopy recv error: %v", recvErr)
-			}
-			break
-		}
-		if resp.GetLastAppendAtNs() != 0 {
-			lastAppendAtNs = resp.GetLastAppendAtNs()
-		}
-	}
-	if lastAppendAtNs == 0 {
-		t.Fatalf("VolumeCopy did not return a lastAppendAtNs timestamp")
-	}
-	t.Logf("VolumeCopy completed, lastAppendAtNs=%d", lastAppendAtNs)
-
-	targetStatusAfterCopy, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus on Rust: %v", err)
-	}
-	t.Logf("Target after copy: dat=%d idx=%d files=%d",
-		targetStatusAfterCopy.GetDatFileSize(), targetStatusAfterCopy.GetIdxFileSize(),
-		targetStatusAfterCopy.GetFileCount())
-
-	if sourceStatus.GetDatFileSize() != targetStatusAfterCopy.GetDatFileSize() {
-		t.Fatalf("dat file size mismatch: source=%d target=%d",
-			sourceStatus.GetDatFileSize(), targetStatusAfterCopy.GetDatFileSize())
-	}
-	if sourceStatus.GetIdxFileSize() != targetStatusAfterCopy.GetIdxFileSize() {
-		t.Fatalf("idx file size mismatch: source=%d target=%d",
-			sourceStatus.GetIdxFileSize(), targetStatusAfterCopy.GetIdxFileSize())
-	}
-
-	// Tail from the copy checkpoint — source is unchanged (deletes happened
-	// before copy), so tailing should not append any data.
-	_, err = rustClient.VolumeTailReceiver(ctx, &volume_server_pb.VolumeTailReceiverRequest{
-		VolumeId:           volumeID,
-		SinceNs:            lastAppendAtNs,
-		IdleTimeoutSeconds: 3,
-		SourceVolumeServer: cluster.VolumeServerAddress(0),
-	})
-	if err != nil {
-		t.Fatalf("VolumeTailReceiver: %v", err)
-	}
-
-	targetStatusAfterTail, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("ReadVolumeFileStatus after tail: %v", err)
-	}
-	if targetStatusAfterTail.GetDatFileSize() != targetStatusAfterCopy.GetDatFileSize() {
-		t.Fatalf("dat grew after tail: before=%d after=%d",
-			targetStatusAfterCopy.GetDatFileSize(), targetStatusAfterTail.GetDatFileSize())
-	}
-	if targetStatusAfterTail.GetIdxFileSize() != targetStatusAfterCopy.GetIdxFileSize() {
-		t.Fatalf("idx grew after tail: before=%d after=%d",
-			targetStatusAfterCopy.GetIdxFileSize(), targetStatusAfterTail.GetIdxFileSize())
-	}
-
-	// Verify surviving files are readable from Rust
-	for _, key := range []uint64{1, 3, 5} {
-		fid := framework.NewFileID(volumeID, key, 0x12340000+uint32(key))
-		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
-		body := framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusOK {
-			t.Fatalf("read surviving file %d from Rust: expected 200, got %d", key, resp.StatusCode)
-		}
-		expected := fmt.Sprintf("file-%d", key)
-		if string(body) != expected {
-			t.Fatalf("data mismatch for file %d: got %q want %q", key, body, expected)
-		}
-	}
-
-	// Verify deleted files return 404
-	for _, key := range []uint64{2, 4} {
-		fid := framework.NewFileID(volumeID, key, 0x12340000+uint32(key))
-		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
-		framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusNotFound {
-			t.Fatalf("read deleted file %d from Rust: expected 404, got %d", key, resp.StatusCode)
-		}
-	}
-}
-
-// TestMixedBalanceFullMoveGoToRust exercises the complete volume balance move
-// flow: mark readonly → copy → tail → verify sizes → delete source.
-// This mirrors the steps in balance_task.go Execute().
-func TestMixedBalanceFullMoveGoToRust(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test in short mode")
-	}
-
-	cluster := framework.StartMixedVolumeCluster(t, matrix.P1(), 1, 1)
-
-	conn0, goClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
-	defer conn0.Close()
-	conn1, rustClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
-	defer conn1.Close()
-
-	httpClient := framework.NewHTTPClient()
-
-	const volumeID = uint32(53)
-	framework.AllocateVolume(t, goClient, volumeID, "")
-
-	// Upload test data
-	for i := uint64(1); i <= 5; i++ {
-		fid := framework.NewFileID(volumeID, i, 0xABCD0000+uint32(i))
-		resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(0), fid, []byte(fmt.Sprintf("balance-move-file-%d", i)))
-		framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusCreated {
-			t.Fatalf("upload file %d: expected 201, got %d", i, resp.StatusCode)
-		}
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
-	defer cancel()
-
-	// Step 1: Copy volume to Rust server (source stays writable so we can
-	// delete after copy to exercise the tail tombstone path)
-	copyStream, err := rustClient.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{
-		VolumeId:       volumeID,
-		SourceDataNode: cluster.VolumeServerAddress(0),
-	})
-	if err != nil {
-		t.Fatalf("VolumeCopy: %v", err)
-	}
-	var lastAppendAtNs uint64
-	for {
-		resp, recvErr := copyStream.Recv()
-		if recvErr != nil {
-			if recvErr != io.EOF {
-				t.Fatalf("VolumeCopy recv: %v", recvErr)
-			}
-			break
-		}
-		if resp.GetLastAppendAtNs() != 0 {
-			lastAppendAtNs = resp.GetLastAppendAtNs()
-		}
-	}
-	t.Logf("Copy done, lastAppendAtNs=%d", lastAppendAtNs)
-
-	// Step 2: Delete file 3 on the source AFTER copy. This creates a
-	// tombstone needle that the tail step must propagate to the Rust server.
-	deleteAndWaitForTombstone(t, httpClient, cluster.VolumeAdminURL(0),
-		framework.NewFileID(volumeID, 3, 0xABCD0003))
-
-	// Step 3: Mark source readonly so no further writes arrive during tail
-	_, err = goClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("mark readonly: %v", err)
-	}
-
-	// Read source status (the reference for post-tail verification)
-	sourceStatus, err := goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("read source status: %v", err)
-	}
-	t.Logf("Source: dat=%d idx=%d files=%d",
-		sourceStatus.GetDatFileSize(), sourceStatus.GetIdxFileSize(),
-		sourceStatus.GetFileCount())
-
-	// Step 4: Tail for updates — this must pick up the delete tombstone
-	_, err = rustClient.VolumeTailReceiver(ctx, &volume_server_pb.VolumeTailReceiverRequest{
-		VolumeId:           volumeID,
-		SinceNs:            lastAppendAtNs,
-		IdleTimeoutSeconds: 5,
-		SourceVolumeServer: cluster.VolumeServerAddress(0),
-	})
-	if err != nil {
-		t.Fatalf("VolumeTailReceiver: %v", err)
-	}
-
-	// Step 5: Verify dat/idx sizes match after tail.
-	// We compare file sizes (byte-level correctness) rather than file_count
-	// because the tail writes the tombstone via the write path (which
-	// increments file_count) while the source used the delete path (which
-	// only increments deletion_count). This is consistent with Go behavior.
-	targetStatus, err := rustClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("read target status: %v", err)
-	}
-	t.Logf("Target: dat=%d idx=%d files=%d",
-		targetStatus.GetDatFileSize(), targetStatus.GetIdxFileSize(),
-		targetStatus.GetFileCount())
-
-	if sourceStatus.GetDatFileSize() != targetStatus.GetDatFileSize() {
-		t.Fatalf("dat size mismatch after tail: source=%d target=%d",
-			sourceStatus.GetDatFileSize(), targetStatus.GetDatFileSize())
-	}
-	if sourceStatus.GetIdxFileSize() != targetStatus.GetIdxFileSize() {
-		t.Fatalf("idx size mismatch after tail: source=%d target=%d",
-			sourceStatus.GetIdxFileSize(), targetStatus.GetIdxFileSize())
-	}
-
-	// Step 6: Delete volume from source
-	_, err = goClient.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{
-		VolumeId: volumeID,
-	})
-	if err != nil {
-		t.Fatalf("delete source volume: %v", err)
-	}
-
-	// Verify source volume is gone
-	_, err = goClient.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
-		VolumeId: volumeID,
-	})
-	if err == nil {
-		t.Fatalf("expected error reading deleted source volume, got nil")
-	}
-	t.Logf("Source volume deleted successfully (error as expected: %v)", err)
-
-	// Verify all surviving data is readable from Rust (the new home)
-	for _, key := range []uint64{1, 2, 4, 5} {
-		fid := framework.NewFileID(volumeID, key, 0xABCD0000+uint32(key))
-		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
-		body := framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusOK {
-			t.Fatalf("read file %d from Rust after move: expected 200, got %d", key, resp.StatusCode)
-		}
-		expected := fmt.Sprintf("balance-move-file-%d", key)
-		if string(body) != expected {
-			t.Fatalf("data mismatch for file %d after move: got %q want %q", key, body, expected)
-		}
-	}
-
-	// Verify deleted file is 404 on Rust (tombstone propagated via tail)
-	{
-		fid := framework.NewFileID(volumeID, 3, 0xABCD0003)
-		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), fid)
-		framework.ReadAllAndClose(t, resp)
-		if resp.StatusCode != http.StatusNotFound {
-			t.Fatalf("deleted file 3 on Rust: expected 404, got %d", resp.StatusCode)
-		}
-	}
-
-	t.Logf("Full balance move completed: volume %d moved from Go to Rust, source purged", volumeID)
-}

weed/admin/dash/admin_data.go

@@ -45,7 +45,6 @@ type ObjectStoreUser struct {
 	SecretKey   string   `json:"secret_key"`
 	Permissions []string `json:"permissions"`
 	PolicyNames []string `json:"policy_names"`
-	IsStatic    bool     `json:"is_static"` // loaded from static config file, not editable
 }
 
 type ObjectStoreUsersData struct {

weed/admin/dash/admin_server.go

@@ -867,24 +867,6 @@ func (s *AdminServer) DeleteS3Bucket(bucketName string) error {
 	})
 }
 
-// IsStaticUser checks if a user is a static identity by loading the
-// configuration from the credential manager and checking the IsStatic flag.
-func (s *AdminServer) IsStaticUser(username string) bool {
-	if s.credentialManager == nil {
-		return false
-	}
-	s3cfg, err := s.credentialManager.LoadConfiguration(context.Background())
-	if err != nil {
-		return false
-	}
-	for _, ident := range s3cfg.Identities {
-		if ident.Name == username {
-			return ident.IsStatic
-		}
-	}
-	return false
-}
-
 // GetObjectStoreUsers retrieves object store users from identity.json
 func (s *AdminServer) GetObjectStoreUsers(ctx context.Context) ([]ObjectStoreUser, error) {
 	if s.credentialManager == nil {
@@ -908,7 +890,6 @@ func (s *AdminServer) GetObjectStoreUsers(ctx context.Context) ([]ObjectStoreUse
 		user := ObjectStoreUser{
 			Username:    identity.Name,
 			Permissions: identity.Actions,
-			IsStatic:    identity.IsStatic,
 		}
 
 		// Set email from account if available

weed/admin/dash/user_management.go

@@ -175,7 +175,7 @@ func (s *AdminServer) GetObjectStoreUserDetails(username string) (*UserDetails,
 
 	ctx := context.Background()
 
-	// Get user using credential manager (resolves static users via filer gRPC)
+	// Get user using credential manager
 	identity, err := s.credentialManager.GetUser(ctx, username)
 	if err != nil {
 		if err == credential.ErrUserNotFound {

weed/admin/handlers/user_handlers.go

@@ -93,11 +93,6 @@ func (h *UserHandlers) UpdateUser(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if h.adminServer.IsStaticUser(username) {
-		writeJSONError(w, http.StatusForbidden, "Cannot modify static user "+username+" (loaded from config file)")
-		return
-	}
-
 	var req dash.UpdateUserRequest
 	if err := decodeJSONBody(newJSONMaxReader(w, r), &req); err != nil {
 		writeJSONError(w, http.StatusBadRequest, "Invalid request: "+err.Error())
@@ -125,11 +120,6 @@ func (h *UserHandlers) DeleteUser(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if h.adminServer.IsStaticUser(username) {
-		writeJSONError(w, http.StatusForbidden, "Cannot delete static user "+username+" (loaded from config file)")
-		return
-	}
-
 	err := h.adminServer.DeleteObjectStoreUser(username)
 	if err != nil {
 		glog.Errorf("Failed to delete user %s: %v", username, err)

weed/admin/view/app/object_store_users.templ

@@ -125,9 +125,6 @@ templ ObjectStoreUsers(data dash.ObjectStoreUsersData) {
                                                 <div class="d-flex align-items-center">
                                                     <i class="fas fa-user me-2 text-muted"></i>
                                                     <strong>{user.Username}</strong>
-                                                    if user.IsStatic {
-                                                        <span class="badge bg-secondary ms-2" title="Loaded from config file (read-only)">static</span>
-                                                    }
                                                 </div>
                                             </td>
                                             <td>{user.Email}</td>
@@ -136,28 +133,24 @@ templ ObjectStoreUsers(data dash.ObjectStoreUsersData) {
                                             </td>
                                             <td>
                                                 <div class="btn-group btn-group-sm" role="group">
-                                                    <button type="button" class="btn btn-outline-info"
+                                                    <button type="button" class="btn btn-outline-info" 
                                                             data-action="show-user-details" data-username={ user.Username }>
                                                         <i class="fas fa-info-circle"></i>
                                                     </button>
-                                                    if !user.IsStatic {
-                                                        <button type="button" class="btn btn-outline-primary"
-                                                                data-action="edit-user" data-username={ user.Username }>
-                                                            <i class="fas fa-edit"></i>
-                                                        </button>
-                                                    }
-                                                    if user.Username != "anonymous" && !user.IsStatic {
+                                                    <button type="button" class="btn btn-outline-primary" 
+                                                            data-action="edit-user" data-username={ user.Username }>
+                                                        <i class="fas fa-edit"></i>
+                                                    </button>
+                                                    if user.Username != "anonymous" {
                                                         <button type="button" class="btn btn-outline-secondary"
                                                                 data-action="manage-access-keys" data-username={ user.Username }>
                                                             <i class="fas fa-key"></i>
                                                         </button>
                                                     }
-                                                    if !user.IsStatic {
-                                                        <button type="button" class="btn btn-outline-danger"
-                                                                data-action="delete-user" data-username={ user.Username }>
-                                                            <i class="fas fa-trash"></i>
-                                                        </button>
-                                                    }
+                                                    <button type="button" class="btn btn-outline-danger" 
+                                                            data-action="delete-user" data-username={ user.Username }>
+                                                        <i class="fas fa-trash"></i>
+                                                    </button>
                                                 </div>
                                             </td>
                                         </tr>

weed/command/admin.go

@@ -30,7 +30,6 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
-	"github.com/seaweedfs/seaweedfs/weed/util/grace"
 )
 
 var (
@@ -49,10 +48,6 @@ type AdminOptions struct {
 	dataDir          *string
 	icebergPort      *int
 	urlPrefix        *string
-	debug            *bool
-	debugPort        *int
-	cpuProfile       *string
-	memProfile       *string
 }
 
 func init() {
@@ -69,10 +64,6 @@ func init() {
 	a.readOnlyPassword = cmdAdmin.Flag.String("readOnlyPassword", "", "read-only user password (optional, for view-only access; requires adminPassword to be set)")
 	a.icebergPort = cmdAdmin.Flag.Int("iceberg.port", 8181, "Iceberg REST Catalog port (0 to hide in UI)")
 	a.urlPrefix = cmdAdmin.Flag.String("urlPrefix", "", "URL path prefix when running behind a reverse proxy under a subdirectory (e.g. /seaweedfs)")
-	a.debug = cmdAdmin.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
-	a.debugPort = cmdAdmin.Flag.Int("debug.port", 6060, "http port for debugging")
-	a.cpuProfile = cmdAdmin.Flag.String("cpuprofile", "", "cpu profile output file")
-	a.memProfile = cmdAdmin.Flag.String("memprofile", "", "memory profile output file")
 }
 
 var cmdAdmin = &Command{
@@ -149,16 +140,6 @@ var cmdAdmin = &Command{
     - All static assets, API endpoints, and navigation links will use the prefix
     - Session cookies are scoped to the prefix path
 
-  Debugging and Profiling:
-    - Use -debug to start a pprof HTTP server for live profiling (localhost only)
-    - Set -debug.port to choose the pprof port (default 6060)
-    - Profiles are accessible at http://127.0.0.1:<debug.port>/debug/pprof/
-    - Use -cpuprofile and -memprofile to write profiles to files on shutdown
-    - WARNING: -debug exposes runtime internals; use only in trusted environments
-    - Examples:
-      weed admin -debug -debug.port=6060 -master="localhost:9333"
-      weed admin -cpuprofile=cpu.prof -memprofile=mem.prof -master="localhost:9333"
-
   Configuration File:
     - The security.toml file is read from ".", "$HOME/.seaweedfs/",
       "/usr/local/etc/seaweedfs/", or "/etc/seaweedfs/", in that order
@@ -168,12 +149,6 @@ var cmdAdmin = &Command{
 }
 
 func runAdmin(cmd *Command, args []string) bool {
-	if *a.debug {
-		grace.StartDebugServer(*a.debugPort)
-	}
-
-	grace.SetupProfiling(*a.cpuProfile, *a.memProfile)
-
 	// Load security configuration
 	util.LoadSecurityConfiguration()
 

weed/command/filer.go

@@ -81,7 +81,6 @@ type FilerOptions struct {
 	exposeDirectoryData       *bool
 	tusBasePath               *string
 	certProvider              certprovider.Provider
-	s3ConfigFile              *string // optional path to static S3 identity config
 }
 
 func init() {
@@ -344,15 +343,6 @@ func (fo *FilerOptions) startFiler() {
 		glog.V(0).Infof("Initialized credential manager: %s", credentialManager.GetStoreName())
 	}
 
-	// Load static S3 identities from config file if specified
-	if fo.s3ConfigFile != nil && *fo.s3ConfigFile != "" {
-		if credentialManager != nil {
-			if err := credentialManager.LoadS3ConfigFile(*fo.s3ConfigFile); err != nil {
-				glog.Warningf("Failed to load S3 config file for static identities: %v", err)
-			}
-		}
-	}
-
 	fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{
 		Masters:                   fo.masters,
 		FilerGroup:                *fo.filerGroup,

weed/command/mini.go

@@ -820,10 +820,6 @@ func runMini(cmd *Command, args []string) bool {
 	miniFilerOptions.disableHttp = miniDisableHttp
 	miniMasterOptions.disableHttp = miniDisableHttp
 
-	// Share the S3 static identity config file with the filer so its
-	// credential manager can also serve static users.
-	miniFilerOptions.s3ConfigFile = miniS3Config
-
 	filerAddress := string(pb.NewServerAddress(*miniIp, *miniFilerOptions.port, *miniFilerOptions.portGrpc))
 	miniS3Options.filer = &filerAddress
 	miniWebDavOptions.filer = &filerAddress

weed/command/server.go

@@ -372,8 +372,6 @@ func runServer(cmd *Command, args []string) bool {
 		} else if *serverIamConfig != "" && *s3Options.iamConfig != *serverIamConfig {
 			glog.V(0).Infof("both -s3.iam.config(%s) and -iam.config(%s) provided; using -s3.iam.config", *s3Options.iamConfig, *serverIamConfig)
 		}
-		// Share the S3 static identity config file with the filer
-		filerOptions.s3ConfigFile = s3Options.config
 		go func() {
 			time.Sleep(2 * time.Second)
 			s3Options.localFilerSocket = filerOptions.localSocket

weed/credential/credential_manager.go

@@ -3,18 +3,14 @@ package credential
 import (
 	"context"
 	"fmt"
-	"os"
 	"strings"
-	"sync"
 
-	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"google.golang.org/grpc"
-	"google.golang.org/protobuf/encoding/protojson"
 )
 
 // FilerAddressSetter is an interface for credential stores that need a dynamic filer address
@@ -25,15 +21,6 @@ type FilerAddressSetter interface {
 // CredentialManager manages user credentials using a configurable store
 type CredentialManager struct {
 	Store CredentialStore
-	// staticMu protects staticIdentities and staticNames, which are written
-	// by SetStaticIdentities (startup + config reload) and read concurrently
-	// by LoadConfiguration, SaveConfiguration, GetStaticUsernames, and IsStaticIdentity.
-	staticMu sync.RWMutex
-	// staticIdentities holds identities loaded from a static config file (-s3.config).
-	// These are included in LoadConfiguration so that listing operations
-	// return all configured identities, not just dynamic ones from the store.
-	staticIdentities []*iam_pb.Identity
-	staticNames      map[string]bool
 }
 
 // NewCredentialManager creates a new credential manager with the specified store
@@ -87,97 +74,13 @@ func (cm *CredentialManager) GetStoreName() string {
 	return ""
 }
 
-// SetStaticIdentities registers identities loaded from a static config file.
-// These identities are included in LoadConfiguration and ListUsers results
-// but are never persisted to the dynamic store.
-func (cm *CredentialManager) SetStaticIdentities(identities []*iam_pb.Identity) {
-	filtered := make([]*iam_pb.Identity, 0, len(identities))
-	names := make(map[string]bool, len(identities))
-	for _, ident := range identities {
-		if ident != nil {
-			filtered = append(filtered, ident)
-			names[ident.Name] = true
-		}
-	}
-	cm.staticMu.Lock()
-	cm.staticIdentities = filtered
-	cm.staticNames = names
-	cm.staticMu.Unlock()
-}
-
-// IsStaticIdentity returns true if the named identity was loaded from static config.
-func (cm *CredentialManager) IsStaticIdentity(name string) bool {
-	cm.staticMu.RLock()
-	defer cm.staticMu.RUnlock()
-	return cm.staticNames[name]
-}
-
-// GetStaticIdentity returns the protobuf identity for a static user, or nil.
-func (cm *CredentialManager) GetStaticIdentity(name string) *iam_pb.Identity {
-	cm.staticMu.RLock()
-	defer cm.staticMu.RUnlock()
-	for _, ident := range cm.staticIdentities {
-		if ident.Name == name {
-			return ident
-		}
-	}
-	return nil
-}
-
-// GetStaticUsernames returns the names of all static identities.
-func (cm *CredentialManager) GetStaticUsernames() []string {
-	cm.staticMu.RLock()
-	defer cm.staticMu.RUnlock()
-	names := make([]string, 0, len(cm.staticIdentities))
-	for _, ident := range cm.staticIdentities {
-		names = append(names, ident.Name)
-	}
-	return names
-}
-
-// LoadConfiguration loads the S3 API configuration from the store and merges
-// in any static identities so that listing operations show all users.
+// LoadConfiguration loads the S3 API configuration
 func (cm *CredentialManager) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) {
-	config, err := cm.Store.LoadConfiguration(ctx)
-	if err != nil {
-		return config, err
-	}
-	// Merge static identities that are not already in the dynamic config
-	cm.staticMu.RLock()
-	staticIdents := cm.staticIdentities
-	cm.staticMu.RUnlock()
-	if len(staticIdents) > 0 {
-		dynamicNames := make(map[string]bool, len(config.Identities))
-		for _, ident := range config.Identities {
-			dynamicNames[ident.Name] = true
-		}
-		for _, si := range staticIdents {
-			if !dynamicNames[si.Name] {
-				config.Identities = append(config.Identities, si)
-			}
-		}
-	}
-	return config, nil
+	return cm.Store.LoadConfiguration(ctx)
 }
 
-// SaveConfiguration saves the S3 API configuration.
-// Static identities are filtered out before saving to the store.
-// The caller's config is not mutated.
+// SaveConfiguration saves the S3 API configuration
 func (cm *CredentialManager) SaveConfiguration(ctx context.Context, config *iam_pb.S3ApiConfiguration) error {
-	cm.staticMu.RLock()
-	staticNames := cm.staticNames
-	cm.staticMu.RUnlock()
-	if len(staticNames) > 0 {
-		var dynamicOnly []*iam_pb.Identity
-		for _, ident := range config.Identities {
-			if !staticNames[ident.Name] {
-				dynamicOnly = append(dynamicOnly, ident)
-			}
-		}
-		configCopy := *config
-		configCopy.Identities = dynamicOnly
-		return cm.Store.SaveConfiguration(ctx, &configCopy)
-	}
 	return cm.Store.SaveConfiguration(ctx, config)
 }
 
@@ -201,12 +104,7 @@ func (cm *CredentialManager) DeleteUser(ctx context.Context, username string) er
 	return cm.Store.DeleteUser(ctx, username)
 }
 
-// ListUsers returns usernames from the dynamic store via cm.Store.ListUsers.
-// On store error the error is returned directly without merging static entries.
-// Static identities (cm.staticIdentities) are NOT included here because
-// internal callers (e.g. DeletePolicy) look up each user in the store and
-// would fail on non-existent static entries. External callers that need the
-// full list should merge GetStaticUsernames separately.
+// ListUsers returns all usernames
 func (cm *CredentialManager) ListUsers(ctx context.Context) ([]string, error) {
 	return cm.Store.ListUsers(ctx)
 }
@@ -271,26 +169,6 @@ func (cm *CredentialManager) UpdatePolicy(ctx context.Context, name string, docu
 	return cm.Store.PutPolicy(ctx, name, document)
 }
 
-// LoadS3ConfigFile reads a static S3 identity config file and registers
-// the identities so they appear in LoadConfiguration and listing results.
-func (cm *CredentialManager) LoadS3ConfigFile(path string) error {
-	content, err := os.ReadFile(path)
-	if err != nil {
-		return fmt.Errorf("read %s: %w", path, err)
-	}
-	config := &iam_pb.S3ApiConfiguration{}
-	opts := protojson.UnmarshalOptions{DiscardUnknown: true, AllowPartial: true}
-	if err := opts.Unmarshal(content, config); err != nil {
-		return fmt.Errorf("parse %s: %w", path, err)
-	}
-	for _, ident := range config.Identities {
-		ident.IsStatic = true
-	}
-	cm.SetStaticIdentities(config.Identities)
-	glog.V(1).Infof("Loaded %d static identities from %s", len(config.Identities), path)
-	return nil
-}
-
 // Shutdown performs cleanup
 func (cm *CredentialManager) Shutdown() {
 	if cm.Store != nil {

weed/filer/filerstore_wrapper.go

@@ -128,6 +128,8 @@ func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefi
 }
 
 func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
+	// Fail fast if the context is already cancelled to prevent orphaned metadata
+	// when the originating request has been abandoned.
 	if err := ctx.Err(); err != nil {
 		return err
 	}
@@ -245,7 +247,7 @@ func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (
 }
 
 func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
-	if err := ctx.Err(); err != nil {
+	if err = ctx.Err(); err != nil {
 		return err
 	}
 	ctx = context.WithoutCancel(ctx)
@@ -276,7 +278,7 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath)
 }
 
 func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
-	if err := ctx.Err(); err != nil {
+	if err = ctx.Err(); err != nil {
 		return err
 	}
 	ctx = context.WithoutCancel(ctx)
@@ -303,7 +305,7 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry
 }
 
 func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
-	if err := ctx.Err(); err != nil {
+	if err = ctx.Err(); err != nil {
 		return err
 	}
 	ctx = context.WithoutCancel(ctx)
@@ -437,7 +439,7 @@ func (fsw *FilerStoreWrapper) Shutdown() {
 }
 
 func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
-	if err := ctx.Err(); err != nil {
+	if err = ctx.Err(); err != nil {
 		return err
 	}
 	ctx = context.WithoutCancel(ctx)
@@ -448,7 +450,7 @@ func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []by
 	return fsw.getDefaultStore().KvGet(ctx, key)
 }
 func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
-	if err := ctx.Err(); err != nil {
+	if err = ctx.Err(); err != nil {
 		return err
 	}
 	ctx = context.WithoutCancel(ctx)

weed/filer/filerstore_wrapper_test.go

@@ -82,7 +82,7 @@ func cancelledCtx() context.Context {
 // expiredCtx returns a context whose deadline has already passed.
 func expiredCtx() context.Context {
 	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
-	cancel() // release resources immediately as it's already expired
+	_ = cancel // already expired, but keep the cancel func from leaking
 	return ctx
 }
 
@@ -198,10 +198,9 @@ func TestFilerStoreWrapperReadOpsSucceedWithCancelledContext(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-// RollbackTransaction must succeed even when the context is cancelled or
-// expired, because it is a cleanup operation called after failures.
+// RollbackTransaction must succeed even when the context is cancelled,
+// because it is a cleanup operation called after failures.
 func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) {
 	wrapper := NewFilerStoreWrapper(newStubFilerStore())
 	assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx()))
-	assert.NoError(t, wrapper.RollbackTransaction(expiredCtx()))
 }

weed/filer/stream.go

@@ -245,66 +245,6 @@ func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclien
 	}, nil
 }
 
-// PrepareStreamContentWithPrefetch is like PrepareStreamContentWithThrottler but uses
-// concurrent chunk prefetching to overlap network I/O. When prefetchAhead > 1, fetch
-// goroutines establish HTTP connections to volume servers ahead of time, streaming data
-// through io.Pipe with minimal memory overhead.
-//
-// prefetchAhead controls the number of chunks fetched concurrently:
-//   - 0 or 1: falls back to sequential fetching (same as PrepareStreamContentWithThrottler)
-//   - 2+: uses pipe-based prefetch pipeline with that many concurrent fetches
-func PrepareStreamContentWithPrefetch(ctx context.Context, masterClient wdclient.HasLookupFileIdFunction, jwtFunc VolumeServerJwtFunction, chunks []*filer_pb.FileChunk, offset int64, size int64, downloadMaxBytesPs int64, prefetchAhead int) (DoStreamContent, error) {
-	if prefetchAhead <= 1 {
-		return PrepareStreamContentWithThrottler(ctx, masterClient, jwtFunc, chunks, offset, size, downloadMaxBytesPs)
-	}
-
-	glog.V(4).InfofCtx(ctx, "prepare to stream content with prefetch=%d for chunks: %d", prefetchAhead, len(chunks))
-	chunkViews := ViewFromChunks(ctx, masterClient.GetLookupFileIdFunction(), chunks, offset, size)
-
-	fileId2Url := make(map[string][]string)
-
-	for x := chunkViews.Front(); x != nil; x = x.Next {
-		chunkView := x.Value
-		var urlStrings []string
-		var err error
-		for _, backoff := range getLookupFileIdBackoffSchedule {
-			if err := ctx.Err(); err != nil {
-				return nil, err
-			}
-			urlStrings, err = masterClient.GetLookupFileIdFunction()(ctx, chunkView.FileId)
-			if err == nil && len(urlStrings) > 0 {
-				break
-			}
-			if err := ctx.Err(); err != nil {
-				return nil, err
-			}
-			glog.V(4).InfofCtx(ctx, "waiting for chunk: %s", chunkView.FileId)
-			timer := time.NewTimer(backoff)
-			select {
-			case <-ctx.Done():
-				if !timer.Stop() {
-					<-timer.C
-				}
-				return nil, ctx.Err()
-			case <-timer.C:
-			}
-		}
-		if err != nil {
-			glog.V(1).InfofCtx(ctx, "operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
-			return nil, err
-		} else if len(urlStrings) == 0 {
-			errUrlNotFound := fmt.Errorf("operation LookupFileId %s failed, err: urls not found", chunkView.FileId)
-			glog.ErrorCtx(ctx, errUrlNotFound)
-			return nil, errUrlNotFound
-		}
-		fileId2Url[chunkView.FileId] = urlStrings
-	}
-
-	return func(writer io.Writer) error {
-		return streamChunksPrefetched(ctx, writer, chunkViews, fileId2Url, jwtFunc, masterClient, offset, size, downloadMaxBytesPs, prefetchAhead)
-	}, nil
-}
-
 func StreamContent(masterClient wdclient.HasLookupFileIdFunction, writer io.Writer, chunks []*filer_pb.FileChunk, offset int64, size int64) error {
 	streamFn, err := PrepareStreamContent(masterClient, JwtForVolumeServer, chunks, offset, size)
 	if err != nil {

weed/filer/stream_benchmark_test.go

@@ -1,317 +0,0 @@
-package filer
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"math/rand"
-	"net/http"
-	"net/http/httptest"
-	"os"
-	"strings"
-	"sync"
-	"testing"
-	"time"
-
-	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
-	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	"github.com/seaweedfs/seaweedfs/weed/wdclient"
-)
-
-func TestMain(m *testing.M) {
-	// Initialize the global HTTP client required by ReadUrlAsStream
-	util_http.InitGlobalHttpClient()
-	os.Exit(m.Run())
-}
-
-// mockMasterClientForBenchmark implements HasLookupFileIdFunction and CacheInvalidator
-type mockMasterClientForBenchmark struct {
-	urls map[string][]string
-}
-
-func (m *mockMasterClientForBenchmark) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType {
-	return func(ctx context.Context, fileId string) ([]string, error) {
-		if urls, ok := m.urls[fileId]; ok {
-			return urls, nil
-		}
-		return nil, fmt.Errorf("fileId %s not found", fileId)
-	}
-}
-
-func (m *mockMasterClientForBenchmark) InvalidateCache(fileId string) {}
-
-// noopJwtFunc returns empty JWT for testing
-func noopJwtFunc(fileId string) string {
-	return ""
-}
-
-// createMockVolumeServer creates an httptest server that serves chunk data
-// with configurable per-request latency to simulate network conditions.
-// The latency is applied once per request (simulating RTT), not per byte.
-func createMockVolumeServer(chunkData map[string][]byte, latency time.Duration) *httptest.Server {
-	var mu sync.RWMutex
-	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		// Simulate network latency (connection setup + RTT)
-		if latency > 0 {
-			time.Sleep(latency)
-		}
-
-		// Extract fileId from path (e.g., "/1,abc123")
-		path := r.URL.Path
-		if strings.HasPrefix(path, "/") {
-			path = path[1:]
-		}
-
-		mu.RLock()
-		data, ok := chunkData[path]
-		mu.RUnlock()
-		if !ok {
-			http.Error(w, "not found", http.StatusNotFound)
-			return
-		}
-
-		// Handle Range header
-		rangeHeader := r.Header.Get("Range")
-		if rangeHeader != "" {
-			var start, end int64
-			fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end)
-			if start >= 0 && end < int64(len(data)) && start <= end {
-				w.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1))
-				w.WriteHeader(http.StatusPartialContent)
-				w.Write(data[start : end+1])
-				return
-			}
-		}
-
-		w.Header().Set("Content-Length", fmt.Sprintf("%d", len(data)))
-		w.WriteHeader(http.StatusOK)
-		w.Write(data)
-	}))
-}
-
-// benchmarkConfig holds parameters for a single benchmark scenario
-type benchmarkConfig struct {
-	numChunks  int
-	chunkSize  int
-	latency    time.Duration
-	prefetch   int // 0 = sequential
-}
-
-func (c benchmarkConfig) name() string {
-	name := fmt.Sprintf("chunks=%d/size=%dKB/latency=%dms",
-		c.numChunks, c.chunkSize/1024, c.latency.Milliseconds())
-	if c.prefetch > 0 {
-		name += fmt.Sprintf("/prefetch=%d", c.prefetch)
-	}
-	return name
-}
-
-// setupBenchmark creates mock infrastructure and returns chunks, master client, and cleanup func
-func setupBenchmark(b *testing.B, cfg benchmarkConfig) ([]*filer_pb.FileChunk, *mockMasterClientForBenchmark, func()) {
-	b.Helper()
-
-	// Generate random chunk data
-	chunkData := make(map[string][]byte, cfg.numChunks)
-	chunks := make([]*filer_pb.FileChunk, cfg.numChunks)
-
-	for i := 0; i < cfg.numChunks; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		data := make([]byte, cfg.chunkSize)
-		rand.Read(data)
-		chunkData[fileId] = data
-
-		chunks[i] = &filer_pb.FileChunk{
-			FileId:       fileId,
-			Offset:       int64(i * cfg.chunkSize),
-			Size:         uint64(cfg.chunkSize),
-			ModifiedTsNs: int64(i),
-			Fid:          &filer_pb.FileId{FileKey: uint64(i)},
-		}
-	}
-
-	// Start mock volume server
-	server := createMockVolumeServer(chunkData, cfg.latency)
-
-	// Build URL map
-	urls := make(map[string][]string, cfg.numChunks)
-	for i := 0; i < cfg.numChunks; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		urls[fileId] = []string{server.URL + "/" + fileId}
-	}
-
-	masterClient := &mockMasterClientForBenchmark{urls: urls}
-	cleanup := func() { server.Close() }
-
-	return chunks, masterClient, cleanup
-}
-
-// runSequentialBenchmark runs the current sequential streaming path
-func runSequentialBenchmark(b *testing.B, cfg benchmarkConfig) {
-	chunks, masterClient, cleanup := setupBenchmark(b, cfg)
-	defer cleanup()
-
-	totalSize := int64(cfg.numChunks * cfg.chunkSize)
-
-	b.ResetTimer()
-	b.SetBytes(totalSize)
-
-	for i := 0; i < b.N; i++ {
-		streamFn, err := PrepareStreamContentWithThrottler(
-			context.Background(),
-			masterClient,
-			noopJwtFunc,
-			chunks,
-			0,
-			totalSize,
-			0, // no throttle
-		)
-		if err != nil {
-			b.Fatal(err)
-		}
-		if err := streamFn(io.Discard); err != nil {
-			b.Fatal(err)
-		}
-	}
-}
-
-// BenchmarkStreamSequential benchmarks the current sequential streaming path.
-// This provides the BEFORE baseline for comparison.
-func BenchmarkStreamSequential(b *testing.B) {
-	configs := []benchmarkConfig{
-		// Pure throughput (no latency)
-		{numChunks: 16, chunkSize: 64 * 1024, latency: 0},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 0},
-		// Moderate latency — shows RTT gap overhead
-		{numChunks: 16, chunkSize: 64 * 1024, latency: 5 * time.Millisecond},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond},
-		// High latency — significant RTT overhead
-		{numChunks: 16, chunkSize: 64 * 1024, latency: 20 * time.Millisecond},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 10 * time.Millisecond},
-	}
-
-	for _, cfg := range configs {
-		b.Run(cfg.name(), func(b *testing.B) {
-			runSequentialBenchmark(b, cfg)
-		})
-	}
-}
-
-// BenchmarkStreamSequentialVerify is a quick functional test that the benchmark
-// infrastructure works correctly — ensures data integrity through the pipeline.
-func BenchmarkStreamSequentialVerify(b *testing.B) {
-	cfg := benchmarkConfig{numChunks: 4, chunkSize: 1024, latency: 0}
-	chunks, masterClient, cleanup := setupBenchmark(b, cfg)
-	defer cleanup()
-
-	totalSize := int64(cfg.numChunks * cfg.chunkSize)
-
-	streamFn, err := PrepareStreamContentWithThrottler(
-		context.Background(),
-		masterClient,
-		noopJwtFunc,
-		chunks,
-		0,
-		totalSize,
-		0,
-	)
-	if err != nil {
-		b.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		b.Fatal(err)
-	}
-
-	if buf.Len() != int(totalSize) {
-		b.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
-	}
-}
-
-// runPrefetchBenchmark runs the new prefetch streaming path
-func runPrefetchBenchmark(b *testing.B, cfg benchmarkConfig) {
-	chunks, masterClient, cleanup := setupBenchmark(b, cfg)
-	defer cleanup()
-
-	totalSize := int64(cfg.numChunks * cfg.chunkSize)
-
-	b.ResetTimer()
-	b.SetBytes(totalSize)
-
-	for i := 0; i < b.N; i++ {
-		streamFn, err := PrepareStreamContentWithPrefetch(
-			context.Background(),
-			masterClient,
-			noopJwtFunc,
-			chunks,
-			0,
-			totalSize,
-			0, // no throttle
-			cfg.prefetch,
-		)
-		if err != nil {
-			b.Fatal(err)
-		}
-		if err := streamFn(io.Discard); err != nil {
-			b.Fatal(err)
-		}
-	}
-}
-
-// BenchmarkStreamPrefetch benchmarks the new prefetch streaming path.
-// Compare against BenchmarkStreamSequential for the AFTER measurement.
-func BenchmarkStreamPrefetch(b *testing.B) {
-	configs := []benchmarkConfig{
-		// Pure throughput (no latency) — should be similar to sequential
-		{numChunks: 16, chunkSize: 64 * 1024, latency: 0, prefetch: 4},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 0, prefetch: 4},
-		// Moderate latency — prefetch should eliminate most RTT overhead
-		{numChunks: 16, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 4},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 4},
-		// High latency — most benefit from prefetch
-		{numChunks: 16, chunkSize: 64 * 1024, latency: 20 * time.Millisecond, prefetch: 4},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 10 * time.Millisecond, prefetch: 4},
-		// Vary prefetch count with moderate latency
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 2},
-		{numChunks: 64, chunkSize: 64 * 1024, latency: 5 * time.Millisecond, prefetch: 8},
-	}
-
-	for _, cfg := range configs {
-		b.Run(cfg.name(), func(b *testing.B) {
-			runPrefetchBenchmark(b, cfg)
-		})
-	}
-}
-
-// BenchmarkStreamPrefetchVerify verifies data integrity through the prefetch pipeline.
-func BenchmarkStreamPrefetchVerify(b *testing.B) {
-	cfg := benchmarkConfig{numChunks: 4, chunkSize: 1024, latency: 0, prefetch: 4}
-	chunks, masterClient, cleanup := setupBenchmark(b, cfg)
-	defer cleanup()
-
-	totalSize := int64(cfg.numChunks * cfg.chunkSize)
-
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		context.Background(),
-		masterClient,
-		noopJwtFunc,
-		chunks,
-		0,
-		totalSize,
-		0,
-		cfg.prefetch,
-	)
-	if err != nil {
-		b.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		b.Fatal(err)
-	}
-
-	if buf.Len() != int(totalSize) {
-		b.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
-	}
-}

weed/filer/stream_prefetch.go

@@ -1,274 +0,0 @@
-package filer
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"sync"
-	"time"
-
-	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/stats"
-	"github.com/seaweedfs/seaweedfs/weed/util"
-	"github.com/seaweedfs/seaweedfs/weed/util/mem"
-	"github.com/seaweedfs/seaweedfs/weed/wdclient"
-)
-
-// chunkPipeResult represents a prefetched chunk streaming through a pipe.
-// The fetch goroutine writes data into the pipeWriter; the consumer reads from pipeReader.
-type chunkPipeResult struct {
-	chunkView  *ChunkView
-	reader     *io.PipeReader
-	fetchErr   error         // final error from fetch goroutine
-	written    int64         // bytes written by fetch goroutine
-	done       chan struct{} // closed when fetch goroutine finishes
-	urlStrings []string      // snapshot of URLs at dispatch time (for retry logic)
-}
-
-// streamChunksPrefetched streams chunks with concurrent prefetch using io.Pipe.
-//
-// For each chunk in file order, a goroutine is launched (bounded by a semaphore)
-// that establishes an HTTP connection to the volume server and streams data through
-// an io.Pipe. The consumer reads from pipes in order, writing to the response.
-//
-// Memory usage is minimal: pipes are synchronous (no buffering), and only one
-// reusable copy buffer is allocated for the consumer.
-func streamChunksPrefetched(
-	ctx context.Context,
-	writer io.Writer,
-	chunkViews *IntervalList[*ChunkView],
-	fileId2Url map[string][]string,
-	jwtFunc VolumeServerJwtFunction,
-	masterClient wdclient.HasLookupFileIdFunction,
-	offset int64,
-	size int64,
-	downloadMaxBytesPs int64,
-	prefetchAhead int,
-) error {
-	downloadThrottler := util.NewWriteThrottler(downloadMaxBytesPs)
-
-	// Create a local cancellable context so the consumer can stop the producer
-	// and all in-flight fetch goroutines on error (e.g., client disconnect).
-	localCtx, localCancel := context.WithCancel(ctx)
-	defer localCancel()
-
-	// Ordered channel: one entry per chunk, in file order.
-	// Capacity = prefetchAhead so the producer can run ahead.
-	// Uses pointer to avoid copying the struct while fetch goroutines write to it.
-	results := make(chan *chunkPipeResult, prefetchAhead)
-
-	// Semaphore to limit concurrent fetch goroutines (and thus HTTP connections).
-	sem := make(chan struct{}, prefetchAhead)
-
-	// Producer: walks chunk list, launches fetch goroutines, sends results in order.
-	// The producer only reads from fileId2Url (populated before streaming starts),
-	// so there is no concurrent map access — the consumer never writes to it.
-	var producerWg sync.WaitGroup
-	producerWg.Add(1)
-	go func() {
-		defer producerWg.Done()
-		defer close(results)
-
-		for x := chunkViews.Front(); x != nil; x = x.Next {
-			chunkView := x.Value
-
-			// Check context before starting new fetch
-			select {
-			case <-localCtx.Done():
-				return
-			default:
-			}
-
-			// Acquire semaphore slot (bounds concurrent HTTP connections)
-			select {
-			case sem <- struct{}{}:
-			case <-localCtx.Done():
-				return
-			}
-
-			pr, pw := io.Pipe()
-			urlStrings := fileId2Url[chunkView.FileId]
-			jwt := jwtFunc(chunkView.FileId)
-
-			result := &chunkPipeResult{
-				chunkView:  chunkView,
-				reader:     pr,
-				done:       make(chan struct{}),
-				urlStrings: urlStrings,
-			}
-
-			// Launch fetch goroutine
-			go func(cv *ChunkView, urls []string, jwt string, pw *io.PipeWriter, res *chunkPipeResult) {
-				defer func() { <-sem }() // release semaphore
-				defer close(res.done)
-
-				written, err := retriedStreamFetchChunkData(
-					localCtx, pw, urls, jwt,
-					cv.CipherKey, cv.IsGzipped, cv.IsFullChunk(),
-					cv.OffsetInChunk, int(cv.ViewSize),
-				)
-				res.written = written
-				res.fetchErr = err
-
-				if err != nil {
-					pw.CloseWithError(err)
-				} else {
-					pw.Close()
-				}
-			}(chunkView, urlStrings, jwt, pw, result)
-
-			// Send result to consumer (blocks if channel full, back-pressuring producer)
-			select {
-			case results <- result:
-			case <-localCtx.Done():
-				// Consumer gone; close the pipe and wait for the fetch goroutine
-				// to finish so we don't leak it (this result was never sent to
-				// the channel, so the drain loop won't handle it).
-				pr.Close()
-				<-result.done
-				return
-			}
-		}
-	}()
-
-	// Consumer: reads from results channel in order, writes to response writer.
-	// Use the SeaweedFS memory pool for the copy buffer to reduce GC pressure.
-	copyBuf := mem.Allocate(256 * 1024)
-	defer mem.Free(copyBuf)
-	remaining := size
-
-	var consumeErr error
-	for result := range results {
-		chunkView := result.chunkView
-
-		// Handle gap before this chunk (zero-fill)
-		if offset < chunkView.ViewOffset {
-			gap := chunkView.ViewOffset - offset
-			remaining -= gap
-			glog.V(4).InfofCtx(ctx, "prefetch zero [%d,%d)", offset, chunkView.ViewOffset)
-			if err := writeZero(writer, gap); err != nil {
-				consumeErr = fmt.Errorf("write zero [%d,%d): %w", offset, chunkView.ViewOffset, err)
-				result.reader.Close()
-				break
-			}
-			offset = chunkView.ViewOffset
-		}
-
-		// Stream chunk data from pipe to response
-		start := time.Now()
-		_, copyErr := io.CopyBuffer(writer, result.reader, copyBuf)
-		result.reader.Close()
-
-		// Wait for fetch goroutine to finish to get final error
-		<-result.done
-
-		// Determine the effective error
-		err := copyErr
-		if err == nil && result.fetchErr != nil && result.written == 0 {
-			err = result.fetchErr
-		}
-
-		// If the fetcher itself failed before writing any data, try cache invalidation
-		// + re-fetch (same as sequential path stream.go:197). We check result.fetchErr
-		// and result.written (not copied) to avoid wrongly retrying when the fetch
-		// succeeded but the response writer failed on the first write.
-		if result.fetchErr != nil && result.written == 0 {
-			if err := localCtx.Err(); err != nil {
-				consumeErr = err
-				break
-			}
-			retryErr := retryWithCacheInvalidation(localCtx, writer, chunkView, result.urlStrings, jwtFunc, masterClient)
-			if retryErr != nil {
-				stats.FilerHandlerCounter.WithLabelValues("chunkDownloadError").Inc()
-				consumeErr = fmt.Errorf("read chunk: %w", retryErr)
-				break
-			}
-			// Retry succeeded
-			err = nil
-		} else if err != nil {
-			if localCtx.Err() != nil {
-				consumeErr = localCtx.Err()
-			} else {
-				stats.FilerHandlerCounter.WithLabelValues("chunkDownloadError").Inc()
-				consumeErr = fmt.Errorf("read chunk: %w", err)
-			}
-			break
-		}
-
-		offset += int64(chunkView.ViewSize)
-		remaining -= int64(chunkView.ViewSize)
-		stats.FilerRequestHistogram.WithLabelValues("chunkDownload").Observe(time.Since(start).Seconds())
-		stats.FilerHandlerCounter.WithLabelValues("chunkDownload").Inc()
-		downloadThrottler.MaybeSlowdown(int64(chunkView.ViewSize))
-	}
-
-	// Cancel the local context to stop the producer and any in-flight fetchers early.
-	// This ensures goroutines don't linger after the consumer exits (e.g., on write error).
-	localCancel()
-
-	// Drain remaining results to close pipes and unblock fetch goroutines
-	for result := range results {
-		result.reader.Close()
-		<-result.done
-	}
-
-	// Wait for producer to finish
-	producerWg.Wait()
-
-	if consumeErr != nil {
-		return consumeErr
-	}
-
-	// Handle trailing zero-fill
-	if remaining > 0 {
-		glog.V(4).InfofCtx(ctx, "prefetch zero [%d,%d)", offset, offset+remaining)
-		if err := writeZero(writer, remaining); err != nil {
-			return fmt.Errorf("write zero [%d,%d): %w", offset, offset+remaining, err)
-		}
-	}
-
-	return nil
-}
-
-// retryWithCacheInvalidation attempts to re-fetch a chunk after invalidating the URL cache.
-// This mirrors the retry logic in PrepareStreamContentWithThrottler's sequential path.
-func retryWithCacheInvalidation(
-	ctx context.Context,
-	writer io.Writer,
-	chunkView *ChunkView,
-	oldUrlStrings []string,
-	jwtFunc VolumeServerJwtFunction,
-	masterClient wdclient.HasLookupFileIdFunction,
-) error {
-	invalidator, ok := masterClient.(CacheInvalidator)
-	if !ok {
-		return fmt.Errorf("read chunk %s failed and no cache invalidator available", chunkView.FileId)
-	}
-
-	glog.V(0).InfofCtx(ctx, "prefetch read chunk %s failed, invalidating cache and retrying", chunkView.FileId)
-	invalidator.InvalidateCache(chunkView.FileId)
-
-	newUrlStrings, lookupErr := masterClient.GetLookupFileIdFunction()(ctx, chunkView.FileId)
-	if lookupErr != nil {
-		glog.WarningfCtx(ctx, "failed to re-lookup chunk %s after cache invalidation: %v", chunkView.FileId, lookupErr)
-		return fmt.Errorf("re-lookup chunk %s: %w", chunkView.FileId, lookupErr)
-	}
-	if len(newUrlStrings) == 0 {
-		glog.WarningfCtx(ctx, "re-lookup for chunk %s returned no locations, skipping retry", chunkView.FileId)
-		return fmt.Errorf("re-lookup chunk %s: no locations", chunkView.FileId)
-	}
-
-	if urlSlicesEqual(oldUrlStrings, newUrlStrings) {
-		glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", chunkView.FileId)
-		return fmt.Errorf("read chunk %s failed, same locations after cache invalidation", chunkView.FileId)
-	}
-
-	glog.V(0).InfofCtx(ctx, "retrying read chunk %s with new locations: %v", chunkView.FileId, newUrlStrings)
-	jwt := jwtFunc(chunkView.FileId)
-	_, err := retriedStreamFetchChunkData(
-		ctx, writer, newUrlStrings, jwt,
-		chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(),
-		chunkView.OffsetInChunk, int(chunkView.ViewSize),
-	)
-	return err
-}

weed/filer/stream_prefetch_test.go

@@ -1,365 +0,0 @@
-package filer
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"math/rand"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	"github.com/seaweedfs/seaweedfs/weed/wdclient"
-)
-
-// testMasterClient implements HasLookupFileIdFunction and CacheInvalidator for tests
-type testMasterClient struct {
-	urls             map[string][]string
-	invalidatedCount int32
-}
-
-func (m *testMasterClient) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType {
-	return func(ctx context.Context, fileId string) ([]string, error) {
-		if urls, ok := m.urls[fileId]; ok {
-			return urls, nil
-		}
-		return nil, fmt.Errorf("fileId %s not found", fileId)
-	}
-}
-
-func (m *testMasterClient) InvalidateCache(fileId string) {
-	atomic.AddInt32(&m.invalidatedCount, 1)
-}
-
-func noopJwt(fileId string) string { return "" }
-
-// createTestServer creates a mock volume server that serves chunk data.
-// Supports Range header for partial chunk reads (exercising OffsetInChunk paths).
-func createTestServer(chunkData map[string][]byte) *httptest.Server {
-	var mu sync.RWMutex
-	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		path := r.URL.Path
-		if strings.HasPrefix(path, "/") {
-			path = path[1:]
-		}
-		mu.RLock()
-		data, ok := chunkData[path]
-		mu.RUnlock()
-		if !ok {
-			http.Error(w, "not found", http.StatusNotFound)
-			return
-		}
-
-		// Handle Range header for partial chunk reads
-		rangeHeader := r.Header.Get("Range")
-		if rangeHeader != "" {
-			var start, end int64
-			if _, err := fmt.Sscanf(rangeHeader, "bytes=%d-%d", &start, &end); err == nil {
-				if start < 0 || end >= int64(len(data)) || start > end {
-					w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", len(data)))
-					w.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
-					return
-				}
-				rangeData := data[start : end+1]
-				w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(data)))
-				w.Header().Set("Content-Length", fmt.Sprintf("%d", len(rangeData)))
-				w.WriteHeader(http.StatusPartialContent)
-				w.Write(rangeData)
-				return
-			}
-		}
-
-		w.Header().Set("Content-Length", fmt.Sprintf("%d", len(data)))
-		w.WriteHeader(http.StatusOK)
-		w.Write(data)
-	}))
-}
-
-// makeChunksAndServer creates N chunks of given size, a mock server, and a master client
-func makeChunksAndServer(t *testing.T, numChunks, chunkSize int) ([]*filer_pb.FileChunk, *testMasterClient, map[string][]byte, func()) {
-	t.Helper()
-
-	chunkData := make(map[string][]byte, numChunks)
-	chunks := make([]*filer_pb.FileChunk, numChunks)
-
-	for i := 0; i < numChunks; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		data := make([]byte, chunkSize)
-		rand.Read(data)
-		chunkData[fileId] = data
-
-		chunks[i] = &filer_pb.FileChunk{
-			FileId:       fileId,
-			Offset:       int64(i * chunkSize),
-			Size:         uint64(chunkSize),
-			ModifiedTsNs: int64(i),
-			Fid:          &filer_pb.FileId{FileKey: uint64(i)},
-		}
-	}
-
-	server := createTestServer(chunkData)
-	urls := make(map[string][]string, numChunks)
-	for i := 0; i < numChunks; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		urls[fileId] = []string{server.URL + "/" + fileId}
-	}
-
-	masterClient := &testMasterClient{urls: urls}
-	return chunks, masterClient, chunkData, func() { server.Close() }
-}
-
-// TestPrefetchInOrderDelivery verifies chunks are written to the output in correct file order
-func TestPrefetchInOrderDelivery(t *testing.T) {
-	chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 8, 4096)
-	defer cleanup()
-
-	totalSize := int64(8 * 4096)
-
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		context.Background(), masterClient, noopJwt,
-		chunks, 0, totalSize, 0, 4,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		t.Fatal(err)
-	}
-
-	// Verify total size
-	if buf.Len() != int(totalSize) {
-		t.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
-	}
-
-	// Verify data matches chunk-by-chunk in order
-	result := buf.Bytes()
-	for i := 0; i < 8; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		expected := chunkData[fileId]
-		got := result[i*4096 : (i+1)*4096]
-		if !bytes.Equal(expected, got) {
-			t.Fatalf("chunk %d (%s) data mismatch at offset %d", i, fileId, i*4096)
-		}
-	}
-}
-
-// TestPrefetchSingleChunk verifies the pipeline works with just one chunk
-func TestPrefetchSingleChunk(t *testing.T) {
-	chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 1, 8192)
-	defer cleanup()
-
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		context.Background(), masterClient, noopJwt,
-		chunks, 0, 8192, 0, 4,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		t.Fatal(err)
-	}
-
-	expected := chunkData["1,0"]
-	if !bytes.Equal(expected, buf.Bytes()) {
-		t.Fatal("single chunk data mismatch")
-	}
-}
-
-// TestPrefetchFallbackToSequential verifies prefetch=1 falls back to sequential path
-func TestPrefetchFallbackToSequential(t *testing.T) {
-	chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 4, 1024)
-	defer cleanup()
-
-	totalSize := int64(4 * 1024)
-
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		context.Background(), masterClient, noopJwt,
-		chunks, 0, totalSize, 0, 1, // prefetch=1 -> sequential
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		t.Fatal(err)
-	}
-
-	if buf.Len() != int(totalSize) {
-		t.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
-	}
-
-	// Verify data order
-	result := buf.Bytes()
-	for i := 0; i < 4; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		expected := chunkData[fileId]
-		got := result[i*1024 : (i+1)*1024]
-		if !bytes.Equal(expected, got) {
-			t.Fatalf("chunk %d data mismatch", i)
-		}
-	}
-}
-
-// TestPrefetchContextCancellation verifies all goroutines clean up on cancellation
-func TestPrefetchContextCancellation(t *testing.T) {
-	// Use a slow server so cancellation happens mid-stream
-	var requestCount int32
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		atomic.AddInt32(&requestCount, 1)
-		// Slow response
-		time.Sleep(100 * time.Millisecond)
-		w.Header().Set("Content-Length", "1024")
-		w.WriteHeader(http.StatusOK)
-		w.Write(make([]byte, 1024))
-	}))
-	defer server.Close()
-
-	numChunks := 16
-	chunks := make([]*filer_pb.FileChunk, numChunks)
-	urls := make(map[string][]string, numChunks)
-	for i := 0; i < numChunks; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		chunks[i] = &filer_pb.FileChunk{
-			FileId: fileId, Offset: int64(i * 1024), Size: 1024,
-			ModifiedTsNs: int64(i), Fid: &filer_pb.FileId{FileKey: uint64(i)},
-		}
-		urls[fileId] = []string{server.URL + "/" + fileId}
-	}
-	masterClient := &testMasterClient{urls: urls}
-
-	// Cancel after a short time
-	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
-	defer cancel()
-
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		ctx, masterClient, noopJwt,
-		chunks, 0, int64(numChunks*1024), 0, 4,
-	)
-	if err != nil {
-		// URL resolution may fail due to cancellation — that's expected
-		return
-	}
-
-	err = streamFn(io.Discard)
-	if err == nil {
-		t.Fatal("expected error from cancelled context")
-	}
-
-	// Verify not all chunks were requested (cancellation stopped early)
-	reqs := atomic.LoadInt32(&requestCount)
-	if reqs >= int32(numChunks) {
-		t.Logf("warning: all %d chunks were requested despite cancellation (got %d)", numChunks, reqs)
-	}
-}
-
-// TestPrefetchRangeRequest verifies prefetch works with offset/size subset
-func TestPrefetchRangeRequest(t *testing.T) {
-	chunks, masterClient, chunkData, cleanup := makeChunksAndServer(t, 8, 4096)
-	defer cleanup()
-
-	// Request only chunks 2-5 (offset=8192, size=16384)
-	offset := int64(2 * 4096)
-	size := int64(4 * 4096)
-
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		context.Background(), masterClient, noopJwt,
-		chunks, offset, size, 0, 4,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		t.Fatal(err)
-	}
-
-	if buf.Len() != int(size) {
-		t.Fatalf("expected %d bytes, got %d", size, buf.Len())
-	}
-
-	// Verify data matches chunks 2-5
-	result := buf.Bytes()
-	for i := 2; i < 6; i++ {
-		fileId := fmt.Sprintf("1,%x", i)
-		expected := chunkData[fileId]
-		start := (i - 2) * 4096
-		got := result[start : start+4096]
-		if !bytes.Equal(expected, got) {
-			t.Fatalf("chunk %d data mismatch in range request", i)
-		}
-	}
-}
-
-// TestPrefetchLargePrefetchCount verifies prefetch > numChunks is handled gracefully
-func TestPrefetchLargePrefetchCount(t *testing.T) {
-	chunks, masterClient, _, cleanup := makeChunksAndServer(t, 3, 1024)
-	defer cleanup()
-
-	totalSize := int64(3 * 1024)
-
-	// prefetch=10 but only 3 chunks — should work fine
-	streamFn, err := PrepareStreamContentWithPrefetch(
-		context.Background(), masterClient, noopJwt,
-		chunks, 0, totalSize, 0, 10,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var buf bytes.Buffer
-	if err := streamFn(&buf); err != nil {
-		t.Fatal(err)
-	}
-
-	if buf.Len() != int(totalSize) {
-		t.Fatalf("expected %d bytes, got %d", totalSize, buf.Len())
-	}
-}
-
-// TestPrefetchConcurrentDownloads verifies multiple concurrent prefetch streams
-func TestPrefetchConcurrentDownloads(t *testing.T) {
-	chunks, masterClient, _, cleanup := makeChunksAndServer(t, 8, 2048)
-	defer cleanup()
-
-	totalSize := int64(8 * 2048)
-
-	var wg sync.WaitGroup
-	errors := make(chan error, 4)
-
-	for i := 0; i < 4; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			streamFn, err := PrepareStreamContentWithPrefetch(
-				context.Background(), masterClient, noopJwt,
-				chunks, 0, totalSize, 0, 4,
-			)
-			if err != nil {
-				errors <- err
-				return
-			}
-			if err := streamFn(io.Discard); err != nil {
-				errors <- err
-			}
-		}()
-	}
-
-	wg.Wait()
-	close(errors)
-
-	for err := range errors {
-		t.Fatalf("concurrent download error: %v", err)
-	}
-}

weed/pb/iam.proto

@@ -186,7 +186,6 @@ message Identity {
     bool disabled = 5;  // User status: false = enabled (default), true = disabled
     repeated string service_account_ids = 6;  // IDs of service accounts owned by this user
     repeated string policy_names = 7;
-    bool is_static = 8;  // Loaded from static config file (read-only, not editable via API)
 }
 
 message Credential {

weed/pb/iam_pb/iam.pb.go

@@ -1399,7 +1399,6 @@ type Identity struct {
 	Disabled          bool                   `protobuf:"varint,5,opt,name=disabled,proto3" json:"disabled,omitempty"`                                             // User status: false = enabled (default), true = disabled
 	ServiceAccountIds []string               `protobuf:"bytes,6,rep,name=service_account_ids,json=serviceAccountIds,proto3" json:"service_account_ids,omitempty"` // IDs of service accounts owned by this user
 	PolicyNames       []string               `protobuf:"bytes,7,rep,name=policy_names,json=policyNames,proto3" json:"policy_names,omitempty"`
-	IsStatic          bool                   `protobuf:"varint,8,opt,name=is_static,json=isStatic,proto3" json:"is_static,omitempty"` // Loaded from static config file (read-only, not editable via API)
 	unknownFields     protoimpl.UnknownFields
 	sizeCache         protoimpl.SizeCache
 }
@@ -1483,13 +1482,6 @@ func (x *Identity) GetPolicyNames() []string {
 	return nil
 }
 
-func (x *Identity) GetIsStatic() bool {
-	if x != nil {
-		return x.IsStatic
-	}
-	return false
-}
-
 type Credential struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	AccessKey     string                 `protobuf:"bytes,1,opt,name=access_key,json=accessKey,proto3" json:"access_key,omitempty"`
@@ -3021,7 +3013,7 @@ const file_iam_proto_rawDesc = "" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x12\x18\n" +
 	"\amembers\x18\x02 \x03(\tR\amembers\x12!\n" +
 	"\fpolicy_names\x18\x03 \x03(\tR\vpolicyNames\x12\x1a\n" +
-	"\bdisabled\x18\x04 \x01(\bR\bdisabled\"\xa5\x02\n" +
+	"\bdisabled\x18\x04 \x01(\bR\bdisabled\"\x88\x02\n" +
 	"\bIdentity\x12\x12\n" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x124\n" +
 	"\vcredentials\x18\x02 \x03(\v2\x12.iam_pb.CredentialR\vcredentials\x12\x18\n" +
@@ -3029,8 +3021,7 @@ const file_iam_proto_rawDesc = "" +
 	"\aaccount\x18\x04 \x01(\v2\x0f.iam_pb.AccountR\aaccount\x12\x1a\n" +
 	"\bdisabled\x18\x05 \x01(\bR\bdisabled\x12.\n" +
 	"\x13service_account_ids\x18\x06 \x03(\tR\x11serviceAccountIds\x12!\n" +
-	"\fpolicy_names\x18\a \x03(\tR\vpolicyNames\x12\x1b\n" +
-	"\tis_static\x18\b \x01(\bR\bisStatic\"b\n" +
+	"\fpolicy_names\x18\a \x03(\tR\vpolicyNames\"b\n" +
 	"\n" +
 	"Credential\x12\x1d\n" +
 	"\n" +

weed/s3api/auth_credentials.go

@@ -295,10 +295,6 @@ func NewIdentityAccessManagementWithStore(option *S3ApiServerOption, filerClient
 	// This serves as an in-memory "static" configuration
 	iam.loadEnvironmentVariableCredentials()
 
-	// Update credential manager with all static identities (file + env vars)
-	// so that listing operations via filer gRPC also include them.
-	iam.updateCredentialManagerStaticIdentities()
-
 	// Determine whether to enable S3 authentication based on configuration
 	// For "weed mini" without any S3 config, default to allowing all access (isAuthEnabled = false)
 	// If any credentials are configured (via file, filer, or env vars), enable authentication
@@ -1073,59 +1069,6 @@ func (iam *IdentityAccessManagement) IsStaticIdentity(identityName string) bool
 	return iam.staticIdentityNames[identityName]
 }
 
-// updateCredentialManagerStaticIdentities syncs the current set of static
-// identities to the credential manager. Call this after any operation that
-// changes static identities (startup, config file reload, etc.).
-func (iam *IdentityAccessManagement) updateCredentialManagerStaticIdentities() {
-	if iam.credentialManager != nil {
-		iam.credentialManager.SetStaticIdentities(iam.GetStaticIdentities())
-	}
-}
-
-// GetStaticIdentities returns protobuf representations of all static identities.
-// This is used to include static identities in listing operations (ListUsers, etc.)
-func (iam *IdentityAccessManagement) GetStaticIdentities() []*iam_pb.Identity {
-	iam.m.RLock()
-	defer iam.m.RUnlock()
-
-	var result []*iam_pb.Identity
-	for _, ident := range iam.identities {
-		if !ident.IsStatic {
-			continue
-		}
-		var policyNames []string
-		if len(ident.PolicyNames) > 0 {
-			policyNames = make([]string, len(ident.PolicyNames))
-			copy(policyNames, ident.PolicyNames)
-		}
-		pbIdent := &iam_pb.Identity{
-			Name:        ident.Name,
-			Disabled:    ident.Disabled,
-			PolicyNames: policyNames,
-			IsStatic:    true,
-		}
-		for _, action := range ident.Actions {
-			pbIdent.Actions = append(pbIdent.Actions, string(action))
-		}
-		for _, cred := range ident.Credentials {
-			pbIdent.Credentials = append(pbIdent.Credentials, &iam_pb.Credential{
-				AccessKey: cred.AccessKey,
-				SecretKey: cred.SecretKey,
-				Status:    cred.Status,
-			})
-		}
-		if ident.Account != nil {
-			pbIdent.Account = &iam_pb.Account{
-				Id:           ident.Account.Id,
-				DisplayName:  ident.Account.DisplayName,
-				EmailAddress: ident.Account.EmailAddress,
-			}
-		}
-		result = append(result, pbIdent)
-	}
-	return result
-}
-
 func (iam *IdentityAccessManagement) lookupByAccessKey(accessKey string) (identity *Identity, cred *Credential, found bool) {
 	iam.m.RLock()
 	defer iam.m.RUnlock()

weed/s3api/filer_multipart.go

@@ -65,12 +65,6 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM
 
 	uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "")
 
-	// Validate checksum algorithm before creating the upload directory
-	_, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r)
-	if checksumErrCode != s3err.ErrNone {
-		return nil, checksumErrCode
-	}
-
 	// Prepare error handling outside callback scope
 	var encryptionError error
 
@@ -101,12 +95,6 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM
 		}
 		s3a.applyMultipartEncryptionConfig(entry, encryptionConfig)
 
-		// Store the requested checksum algorithm so CompleteMultipartUpload can compute
-		// a composite checksum from per-part checksums
-		if checksumHeaderName != "" {
-			entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(checksumHeaderName)
-		}
-
 		// Extract and store object lock metadata from request headers
 		// This ensures object lock settings from create_multipart_upload are preserved
 		if err := s3a.extractObjectLockMetadataFromRequest(r, entry); err != nil {
@@ -141,11 +129,6 @@ type CompleteMultipartUploadResult struct {
 	Bucket   *string  `xml:"Bucket,omitempty"`
 	Key      *string  `xml:"Key,omitempty"`
 	ETag     *string  `xml:"ETag,omitempty"`
-
-	// Checksum fields — returned as HTTP response headers, not in the XML body
-	ChecksumHeaderName string `xml:"-"`
-	ChecksumValue      string `xml:"-"`
-
 	// VersionId is NOT included in XML body - it should only be in x-amz-version-id HTTP header
 
 	// Store the VersionId internally for setting HTTP header, but don't marshal to XML
@@ -190,17 +173,15 @@ type multipartPartBoundary struct {
 }
 
 type multipartCompletionState struct {
-	deleteEntries      []*filer_pb.Entry
-	partEntries        map[int][]*filer_pb.Entry
-	pentry             *filer_pb.Entry
-	mime               string
-	finalParts         []*filer_pb.FileChunk
-	offset             int64
-	partBoundaries     []multipartPartBoundary
-	multipartETag      string
-	entityWithTtl      bool
-	checksumHeaderName string // e.g. "X-Amz-Checksum-Crc32", empty if no checksum
-	checksumValue      string // composite base64 checksum with "-N" suffix
+	deleteEntries  []*filer_pb.Entry
+	partEntries    map[int][]*filer_pb.Entry
+	pentry         *filer_pb.Entry
+	mime           string
+	finalParts     []*filer_pb.FileChunk
+	offset         int64
+	partBoundaries []multipartPartBoundary
+	multipartETag  string
+	entityWithTtl  bool
 }
 
 func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult {
@@ -365,36 +346,16 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
 		}
 	}
 
-	// Compute composite checksum from per-part checksums if the upload
-	// was initiated with a checksum algorithm (stored in upload dir entry)
-	checksumHeaderName := ""
-	checksumValue := ""
-	if pentry.Extended != nil {
-		if algoName, ok := pentry.Extended[s3_constants.ExtChecksumAlgorithm]; ok {
-			checksumHeaderName = string(algoName)
-		}
-	}
-	if checksumHeaderName != "" {
-		var checksumErr error
-		checksumValue, checksumErr = computeCompositeChecksum(checksumHeaderName, partEntries, completedPartNumbers)
-		if checksumErr != nil {
-			glog.Errorf("completeMultipartUpload: composite checksum computation failed: %v", checksumErr)
-			return nil, nil, s3err.ErrInvalidPart
-		}
-	}
-
 	return &multipartCompletionState{
-		deleteEntries:      deleteEntries,
-		partEntries:        partEntries,
-		pentry:             pentry,
-		mime:               mime,
-		finalParts:         finalParts,
-		offset:             offset,
-		partBoundaries:     partBoundaries,
-		multipartETag:      calculateMultipartETag(partEntries, completedPartNumbers),
-		entityWithTtl:      entityWithTtl,
-		checksumHeaderName: checksumHeaderName,
-		checksumValue:      checksumValue,
+		deleteEntries:  deleteEntries,
+		partEntries:    partEntries,
+		pentry:         pentry,
+		mime:           mime,
+		finalParts:     finalParts,
+		offset:         offset,
+		partBoundaries: partBoundaries,
+		multipartETag:  calculateMultipartETag(partEntries, completedPartNumbers),
+		entityWithTtl:  entityWithTtl,
 	}, nil, s3err.ErrNone
 }
 
@@ -481,11 +442,6 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 
 				// Persist ETag to ensure subsequent HEAD/GET uses the same value
 				versionEntry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
-				// Store composite checksum if computed from per-part checksums
-				if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
-					versionEntry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
-					versionEntry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
-				}
 
 				// Preserve ALL SSE metadata from the first part (if any)
 				// SSE metadata is stored in individual parts, not the upload directory
@@ -534,13 +490,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 			// For versioned buckets, all content is stored in .versions directory
 			// The latest version information is tracked in the .versions directory metadata
 			output = &CompleteMultipartUploadResult{
-				Location:           aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
-				Bucket:             input.Bucket,
-				ETag:               aws.String(etagQuote),
-				Key:                objectKey(input.Key),
-				VersionId:          aws.String(versionId),
-				ChecksumHeaderName: completionState.checksumHeaderName,
-				ChecksumValue:      completionState.checksumValue,
+				Location:  aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
+				Bucket:    input.Bucket,
+				ETag:      aws.String(etagQuote),
+				Key:       objectKey(input.Key),
+				VersionId: aws.String(versionId),
 			}
 			return s3err.ErrNone
 		}
@@ -580,11 +534,6 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 				}
 				// Persist ETag to ensure subsequent HEAD/GET uses the same value
 				entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
-				// Store composite checksum if computed from per-part checksums
-				if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
-					entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
-					entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
-				}
 				if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
 					entry.Attributes.Mime = completionState.pentry.Attributes.Mime
 				} else if completionState.mime != "" {
@@ -598,12 +547,10 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 
 			// Note: Suspended versioning should NOT return VersionId field according to AWS S3 spec
 			output = &CompleteMultipartUploadResult{
-				Location:           aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
-				Bucket:             input.Bucket,
-				ETag:               aws.String(etagQuote),
-				Key:                objectKey(input.Key),
-				ChecksumHeaderName: completionState.checksumHeaderName,
-				ChecksumValue:      completionState.checksumValue,
+				Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
+				Bucket:   input.Bucket,
+				ETag:     aws.String(etagQuote),
+				Key:      objectKey(input.Key),
 				// VersionId field intentionally omitted for suspended versioning
 			}
 			return s3err.ErrNone
@@ -642,11 +589,6 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 			}
 			// Persist ETag to ensure subsequent HEAD/GET uses the same value
 			entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
-			// Store composite checksum if computed from per-part checksums
-			if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
-				entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
-				entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
-			}
 			if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
 				entry.Attributes.Mime = completionState.pentry.Attributes.Mime
 			} else if completionState.mime != "" {
@@ -664,12 +606,10 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 
 		// For non-versioned buckets, return response without VersionId
 		output = &CompleteMultipartUploadResult{
-			Location:           aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
-			Bucket:             input.Bucket,
-			ETag:               aws.String(etagQuote),
-			Key:                objectKey(input.Key),
-			ChecksumHeaderName: completionState.checksumHeaderName,
-			ChecksumValue:      completionState.checksumValue,
+			Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
+			Bucket:   input.Bucket,
+			ETag:     aws.String(etagQuote),
+			Key:      objectKey(input.Key),
 		}
 		return s3err.ErrNone
 	})
@@ -1090,72 +1030,6 @@ func calculateMultipartETag(partEntries map[int][]*filer_pb.Entry, completedPart
 	return fmt.Sprintf("%x-%d", md5.Sum(etags), len(completedPartNumbers))
 }
 
-// computeCompositeChecksum computes a composite checksum from per-part checksums.
-// It concatenates the raw (decoded) per-part checksums, hashes the result with the
-// same algorithm, and returns the value as "base64-N" where N is the part count.
-// This follows the AWS S3 multipart checksum specification.
-// Returns an error if a part is missing its checksum (the upload was initiated with
-// a checksum algorithm, so all parts must have been uploaded with checksums).
-func computeCompositeChecksum(checksumHeaderName string, partEntries map[int][]*filer_pb.Entry, completedPartNumbers []int) (string, error) {
-	// Determine the algorithm from the header name
-	algo := checksumAlgorithmFromHeaderName(checksumHeaderName)
-	if algo == ChecksumAlgorithmNone {
-		return "", fmt.Errorf("unknown checksum algorithm for header %q", checksumHeaderName)
-	}
-
-	// Collect raw per-part checksums
-	var combined []byte
-	for _, partNumber := range completedPartNumbers {
-		entries, ok := partEntries[partNumber]
-		if !ok || len(entries) == 0 {
-			return "", fmt.Errorf("part %d not found", partNumber)
-		}
-		if len(entries) > 1 {
-			sortEntriesByLatestChunk(entries)
-		}
-		entry := entries[0]
-		if entry.Extended == nil {
-			return "", fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName)
-		}
-		// Validate the part's checksum algorithm matches the upload's expected algorithm
-		partAlgo, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm]
-		if !ok || len(partAlgo) == 0 {
-			return "", fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName)
-		}
-		if string(partAlgo) != checksumHeaderName {
-			return "", fmt.Errorf("part %d checksum algorithm mismatch: upload expects %s but part has %s", partNumber, checksumHeaderName, string(partAlgo))
-		}
-		partChecksumB64, ok := entry.Extended[s3_constants.ExtChecksumValue]
-		if !ok || len(partChecksumB64) == 0 {
-			return "", fmt.Errorf("part %d missing checksum value: upload initiated with %s but part has no checksum value", partNumber, checksumHeaderName)
-		}
-		raw, err := base64.StdEncoding.DecodeString(string(partChecksumB64))
-		if err != nil {
-			return "", fmt.Errorf("part %d has invalid checksum encoding: %w", partNumber, err)
-		}
-		combined = append(combined, raw...)
-	}
-
-	// Hash the concatenated raw checksums
-	h := getCheckSumWriter(algo)
-	if h == nil {
-		return "", fmt.Errorf("failed to create hash writer for %s", checksumHeaderName)
-	}
-	h.Write(combined)
-	compositeRaw := h.Sum(nil)
-	return fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(compositeRaw), len(completedPartNumbers)), nil
-}
-
-// checksumAlgorithmFromHeaderName maps a canonical header name back to its algorithm.
-func checksumAlgorithmFromHeaderName(headerName string) ChecksumAlgorithm {
-	for _, entry := range checksumHeaders {
-		if entry.name == headerName {
-			return entry.alg
-		}
-	}
-	return ChecksumAlgorithmNone
-}
-
 func getEtagFromEntry(entry *filer_pb.Entry) string {
 	if entry.Extended != nil {
 		if etagBytes, ok := entry.Extended[s3_constants.ExtETagKey]; ok {

weed/s3api/s3_constants/extend_key.go

@@ -20,10 +20,6 @@ const (
 	ExtLatestVersionIsDeleteMarker = "Seaweed-X-Amz-Latest-Version-Is-Delete-Marker"
 	ExtMultipartObjectKey          = "key"
 
-	// S3 checksum storage keys (use x-seaweedfs- prefix to avoid leaking in generic header loop)
-	ExtChecksumAlgorithm = "x-seaweedfs-checksum-algorithm"
-	ExtChecksumValue     = "x-seaweedfs-checksum-value"
-
 	// Bucket Policy
 	ExtBucketPolicyKey = "Seaweed-X-Amz-Bucket-Policy"
 

weed/s3api/s3_constants/header.go

@@ -76,16 +76,6 @@ const (
 	AmzObjectLockRetainUntilDate = "X-Amz-Object-Lock-Retain-Until-Date"
 	AmzObjectLockLegalHold       = "X-Amz-Object-Lock-Legal-Hold"
 
-	// S3 checksum headers
-	AmzChecksumAlgorithm = "X-Amz-Checksum-Algorithm"
-	AmzChecksumCRC32     = "X-Amz-Checksum-Crc32"
-	AmzChecksumCRC32C    = "X-Amz-Checksum-Crc32c"
-	AmzChecksumCRC64NVME = "X-Amz-Checksum-Crc64nvme"
-	AmzChecksumSHA1      = "X-Amz-Checksum-Sha1"
-	AmzChecksumSHA256    = "X-Amz-Checksum-Sha256"
-	AmzTrailer           = "X-Amz-Trailer"
-	AmzSdkChecksumAlgorithm = "X-Amz-Sdk-Checksum-Algorithm"
-
 	// S3 conditional headers
 	IfMatch           = "If-Match"
 	IfNoneMatch       = "If-None-Match"

weed/s3api/s3api_embedded_iam.go

@@ -215,7 +215,6 @@ func (e *EmbeddedIamApi) writeIamErrorResponse(w http.ResponseWriter, r *http.Re
 }
 
 // GetS3ApiConfiguration loads the S3 API configuration from the credential manager.
-// The credential manager automatically includes static identities in the result.
 func (e *EmbeddedIamApi) GetS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) error {
 	if e.getS3ApiConfigurationFunc != nil {
 		return e.getS3ApiConfigurationFunc(s3cfg)
@@ -229,7 +228,6 @@ func (e *EmbeddedIamApi) GetS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration)
 }
 
 // PutS3ApiConfiguration saves the S3 API configuration to the credential manager.
-// The credential manager automatically filters out static identities before saving.
 func (e *EmbeddedIamApi) PutS3ApiConfiguration(s3cfg *iam_pb.S3ApiConfiguration) error {
 	if e.putS3ApiConfigurationFunc != nil {
 		return e.putS3ApiConfigurationFunc(s3cfg)

weed/s3api/s3api_object_handlers.go

@@ -1053,7 +1053,7 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 	// Prepare streaming function with simple master client wrapper
 	tStreamPrep := time.Now()
 	// Use filerClient directly (not wrapped) so it can support cache invalidation
-	streamFn, err := filer.PrepareStreamContentWithPrefetch(
+	streamFn, err := filer.PrepareStreamContentWithThrottler(
 		ctx,
 		s3a.filerClient,
 		filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally)
@@ -1061,7 +1061,6 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		offset,
 		size,
 		0, // no throttling
-		4, // prefetch 4 chunks ahead for overlapped fetching
 	)
 	streamPrepTime = time.Since(tStreamPrep)
 	if err != nil {
@@ -1929,7 +1928,7 @@ func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry
 	}
 
 	// Create streaming reader - use filerClient directly for cache invalidation support
-	streamFn, err := filer.PrepareStreamContentWithPrefetch(
+	streamFn, err := filer.PrepareStreamContentWithThrottler(
 		ctx,
 		s3a.filerClient,
 		filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally)
@@ -1937,7 +1936,6 @@ func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry
 		0,
 		totalSize,
 		0,
-		4, // prefetch 4 chunks ahead for overlapped fetching
 	)
 	if err != nil {
 		return nil, err
@@ -2070,21 +2068,6 @@ func (s3a *S3ApiServer) setResponseHeaders(w http.ResponseWriter, r *http.Reques
 		}
 	}
 
-	// Set checksum header if stored in metadata, but only when:
-	// 1. The request contains "x-amz-checksum-mode: ENABLED" (per AWS S3 spec)
-	// 2. The request is NOT a ranged GET (Range header absent)
-	//    The stored checksum covers the full object; returning it for partial
-	//    responses causes SDK checksum validation failures.
-	if r != nil && r.Header.Get("X-Amz-Checksum-Mode") == "ENABLED" && r.Header.Get("Range") == "" {
-		if entry.Extended != nil {
-			if algoName, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm]; ok {
-				if checksumVal, ok := entry.Extended[s3_constants.ExtChecksumValue]; ok {
-					w.Header().Set(string(algoName), string(checksumVal))
-				}
-			}
-		}
-	}
-
 	// Apply S3 passthrough headers from query parameters
 	// AWS S3 supports overriding response headers via query parameters like:
 	// ?response-cache-control=no-cache&response-content-type=application/json

weed/s3api/s3api_object_handlers_multipart.go

@@ -180,11 +180,6 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
 		w.Header().Set("x-amz-version-id", *response.VersionId)
 	}
 
-	// Set composite checksum header if present
-	if response.ChecksumHeaderName != "" && response.ChecksumValue != "" {
-		w.Header().Set(response.ChecksumHeaderName, response.ChecksumValue)
-	}
-
 	stats_collect.RecordBucketActiveTime(bucket)
 	stats_collect.S3UploadedObjectsCounter.WithLabelValues(bucket).Inc()
 

weed/s3api/s3api_object_handlers_put.go

@@ -8,7 +8,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"hash"
 	"io"
 	"net/http"
 	"net/url"
@@ -72,9 +71,6 @@ type SSEResponseMetadata struct {
 	SSEType          string
 	KMSKeyID         string
 	BucketKeyEnabled bool
-	// Checksum fields for S3 additional checksum support
-	ChecksumHeaderName string // e.g. "X-Amz-Checksum-Sha256"
-	ChecksumValue      string // base64-encoded checksum value
 }
 
 func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
@@ -362,19 +358,6 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	plaintextHash := md5.New()
 	dataReader = io.TeeReader(dataReader, plaintextHash)
 
-	// Detect and set up additional checksum computation (S3 checksum algorithm support)
-	checksumAlgo, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r)
-	if checksumErrCode != s3err.ErrNone {
-		return "", checksumErrCode, SSEResponseMetadata{}
-	}
-	var checksumHash hash.Hash
-	if checksumAlgo != ChecksumAlgorithmNone {
-		checksumHash = getCheckSumWriter(checksumAlgo)
-		if checksumHash != nil {
-			dataReader = io.TeeReader(dataReader, checksumHash)
-		}
-	}
-
 	// Handle all SSE encryption types in a unified manner
 	sseResult, sseErrorCode := s3a.handleAllSSEEncryption(r, dataReader, partOffset)
 	if sseErrorCode != s3err.ErrNone {
@@ -651,24 +634,6 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	// Store ETag in Extended attribute for future retrieval (e.g. multipart parts)
 	entry.Extended[s3_constants.ExtETagKey] = []byte(etag)
 
-	// Store additional checksum if one was computed
-	checksumBase64 := ""
-	if checksumHash != nil && checksumHeaderName != "" {
-		checksumBase64 = base64.StdEncoding.EncodeToString(checksumHash.Sum(nil))
-		// Verify against client-provided checksum if present in request headers
-		// (non-chunked uploads send the value directly; chunked uploads validate in the reader)
-		if expectedChecksum := r.Header.Get(checksumHeaderName); expectedChecksum != "" {
-			if expectedChecksum != checksumBase64 {
-				glog.Warningf("putToFiler: checksum mismatch for %s: expected %s, got %s", checksumHeaderName, expectedChecksum, checksumBase64)
-				s3a.deleteOrphanedChunks(chunkResult.FileChunks)
-				return "", s3err.ErrBadDigest, SSEResponseMetadata{}
-			}
-		}
-		entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(checksumHeaderName)
-		entry.Extended[s3_constants.ExtChecksumValue] = []byte(checksumBase64)
-		glog.V(3).Infof("putToFiler: stored checksum %s=%s for %s", checksumHeaderName, checksumBase64, filePath)
-	}
-
 	// Set object owner according to bucket ownership settings.
 	s3a.setObjectOwnerFromRequest(r, bucket, entry)
 
@@ -837,9 +802,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 
 	// Build SSE response metadata with encryption details
 	responseMetadata := SSEResponseMetadata{
-		SSEType:            sseType,
-		ChecksumHeaderName: checksumHeaderName,
-		ChecksumValue:      checksumBase64,
+		SSEType: sseType,
 	}
 
 	// For SSE-KMS, include key ID and bucket-key-enabled flag from stored metadata
@@ -853,93 +816,6 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	return etag, s3err.ErrNone, responseMetadata
 }
 
-// checksumAlgorithmMapping maps algorithm name strings to their enum and header name.
-var checksumAlgorithmMapping = map[string]struct {
-	alg  ChecksumAlgorithm
-	name string
-}{
-	"CRC32":    {ChecksumAlgorithmCRC32, s3_constants.AmzChecksumCRC32},
-	"CRC32C":   {ChecksumAlgorithmCRC32C, s3_constants.AmzChecksumCRC32C},
-	"CRC64NVME": {ChecksumAlgorithmCRC64NVMe, s3_constants.AmzChecksumCRC64NVME},
-	"SHA1":     {ChecksumAlgorithmSHA1, s3_constants.AmzChecksumSHA1},
-	"SHA256":   {ChecksumAlgorithmSHA256, s3_constants.AmzChecksumSHA256},
-}
-
-// trailerToChecksumAlgorithm maps trailer header names to their algorithm and canonical header name.
-var trailerToChecksumAlgorithm = map[string]struct {
-	alg  ChecksumAlgorithm
-	name string
-}{
-	"x-amz-checksum-crc32":    {ChecksumAlgorithmCRC32, s3_constants.AmzChecksumCRC32},
-	"x-amz-checksum-crc32c":   {ChecksumAlgorithmCRC32C, s3_constants.AmzChecksumCRC32C},
-	"x-amz-checksum-crc64nvme": {ChecksumAlgorithmCRC64NVMe, s3_constants.AmzChecksumCRC64NVME},
-	"x-amz-checksum-sha1":     {ChecksumAlgorithmSHA1, s3_constants.AmzChecksumSHA1},
-	"x-amz-checksum-sha256":   {ChecksumAlgorithmSHA256, s3_constants.AmzChecksumSHA256},
-}
-
-// checksumHeaders is the ordered list of individual checksum headers to check.
-// Using a slice ensures deterministic selection order.
-var checksumHeaders = []struct {
-	header string
-	alg    ChecksumAlgorithm
-	name   string
-}{
-	{s3_constants.AmzChecksumCRC32, ChecksumAlgorithmCRC32, s3_constants.AmzChecksumCRC32},
-	{s3_constants.AmzChecksumCRC32C, ChecksumAlgorithmCRC32C, s3_constants.AmzChecksumCRC32C},
-	{s3_constants.AmzChecksumCRC64NVME, ChecksumAlgorithmCRC64NVMe, s3_constants.AmzChecksumCRC64NVME},
-	{s3_constants.AmzChecksumSHA1, ChecksumAlgorithmSHA1, s3_constants.AmzChecksumSHA1},
-	{s3_constants.AmzChecksumSHA256, ChecksumAlgorithmSHA256, s3_constants.AmzChecksumSHA256},
-}
-
-// detectRequestedChecksumAlgorithm detects the checksum algorithm requested by the client.
-// It checks the x-amz-sdk-checksum-algorithm header, x-amz-checksum-algorithm header,
-// x-amz-trailer header (including comma-separated values), and individual x-amz-checksum-*
-// headers. Returns the algorithm enum, the canonical HTTP header name, and an error code
-// if an unsupported algorithm is specified.
-func detectRequestedChecksumAlgorithm(r *http.Request) (ChecksumAlgorithm, string, s3err.ErrorCode) {
-	// Check x-amz-sdk-checksum-algorithm (set by AWS SDKs)
-	if algo := r.Header.Get(s3_constants.AmzSdkChecksumAlgorithm); algo != "" {
-		if m, ok := checksumAlgorithmMapping[strings.ToUpper(algo)]; ok {
-			return m.alg, m.name, s3err.ErrNone
-		}
-		glog.Warningf("unsupported checksum algorithm in %s: %q", s3_constants.AmzSdkChecksumAlgorithm, algo)
-		return ChecksumAlgorithmNone, "", s3err.ErrInvalidRequest
-	}
-
-	// Check x-amz-checksum-algorithm header
-	if algo := r.Header.Get(s3_constants.AmzChecksumAlgorithm); algo != "" {
-		if m, ok := checksumAlgorithmMapping[strings.ToUpper(algo)]; ok {
-			return m.alg, m.name, s3err.ErrNone
-		}
-		glog.Warningf("unsupported checksum algorithm in %s: %q", s3_constants.AmzChecksumAlgorithm, algo)
-		return ChecksumAlgorithmNone, "", s3err.ErrInvalidRequest
-	}
-
-	// Check x-amz-trailer header (used by chunked uploads, may be comma-separated)
-	if trailer := r.Header.Get(s3_constants.AmzTrailer); trailer != "" {
-		for _, part := range strings.Split(trailer, ",") {
-			part = strings.TrimSpace(strings.ToLower(part))
-			if part == "" {
-				continue
-			}
-			if m, ok := trailerToChecksumAlgorithm[part]; ok {
-				return m.alg, m.name, s3err.ErrNone
-			}
-			// Non-checksum trailers (e.g. x-amz-server-side-encryption) are fine — skip them
-		}
-	}
-
-	// Check individual checksum headers (non-chunked uploads send the value directly)
-	// Uses ordered slice for deterministic selection
-	for _, entry := range checksumHeaders {
-		if r.Header.Get(entry.header) != "" {
-			return entry.alg, entry.name, s3err.ErrNone
-		}
-	}
-
-	return ChecksumAlgorithmNone, "", s3err.ErrNone
-}
-
 const defaultFileMode = uint32(0660)
 
 // resolveFileMode determines the file permission mode for an S3 upload.
@@ -1007,11 +883,6 @@ func (s3a *S3ApiServer) setSSEResponseHeaders(w http.ResponseWriter, r *http.Req
 			w.Header().Set(s3_constants.AmzServerSideEncryptionBucketKeyEnabled, "true")
 		}
 	}
-
-	// Set checksum response header if a checksum was computed
-	if sseMetadata.ChecksumHeaderName != "" && sseMetadata.ChecksumValue != "" {
-		w.Header().Set(sseMetadata.ChecksumHeaderName, sseMetadata.ChecksumValue)
-	}
 }
 
 func filerErrorToS3Error(err error) s3err.ErrorCode {

weed/s3api/s3api_server.go

@@ -276,7 +276,6 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 				glog.Errorf("fail to load config file %s: %v", option.Config, err)
 			} else {
 				glog.V(1).Infof("Loaded %d identities from config file %s", len(s3ApiServer.iam.identities), option.Config)
-				s3ApiServer.iam.updateCredentialManagerStaticIdentities()
 			}
 		})
 	}

weed/server/filer_server.go

@@ -82,7 +82,6 @@ type FilerOption struct {
 	AllowedOrigins            []string
 	ExposeDirectoryData       bool
 	TusBasePath               string
-	S3ConfigFile              string // optional path to static S3 identity config file
 	CredentialManager         *credential.CredentialManager
 }
 

weed/server/filer_server_handlers_iam_grpc.go

@@ -101,10 +101,6 @@ func (s *IamGrpcServer) GetUser(ctx context.Context, req *iam_pb.GetUserRequest)
 	identity, err := s.credentialManager.GetUser(ctx, req.Username)
 	if err != nil {
 		if err == credential.ErrUserNotFound {
-			// Fall back to static identities (loaded from -s3.config file)
-			if si := s.credentialManager.GetStaticIdentity(req.Username); si != nil {
-				return &iam_pb.GetUserResponse{Identity: si}, nil
-			}
 			return nil, status.Errorf(codes.NotFound, "user %s not found", req.Username)
 		}
 		glog.Errorf("Failed to get user %s: %v", req.Username, err)
@@ -170,20 +166,6 @@ func (s *IamGrpcServer) ListUsers(ctx context.Context, req *iam_pb.ListUsersRequ
 		return nil, err
 	}
 
-	// Merge static identities (from -s3.config file) into the result
-	staticNames := s.credentialManager.GetStaticUsernames()
-	if len(staticNames) > 0 {
-		dynamicSet := make(map[string]bool, len(usernames))
-		for _, name := range usernames {
-			dynamicSet[name] = true
-		}
-		for _, name := range staticNames {
-			if !dynamicSet[name] {
-				usernames = append(usernames, name)
-			}
-		}
-	}
-
 	return &iam_pb.ListUsersResponse{
 		Usernames: usernames,
 	}, nil

weed/server/filer_server_handlers_read.go

@@ -223,7 +223,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		// Matches S3 API behavior. Request context (ctx) is used for metadata operations above.
 		streamCtx, streamCancel := context.WithCancel(context.WithoutCancel(ctx))
 
-		streamFn, err := filer.PrepareStreamContentWithPrefetch(streamCtx, fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs, 4)
+		streamFn, err := filer.PrepareStreamContentWithThrottler(streamCtx, fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs)
 		if err != nil {
 			streamCancel()
 			stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()

weed/util/grace/pprof.go

@@ -15,8 +15,8 @@ import (
 // The server runs in a goroutine and serves pprof endpoints at /debug/pprof/*.
 func StartDebugServer(debugPort int) {
 	go func() {
-		addr := fmt.Sprintf("127.0.0.1:%d", debugPort)
-		glog.V(0).Infof("Starting debug server for pprof at http://%s/debug/pprof/", addr)
+		addr := fmt.Sprintf(":%d", debugPort)
+		glog.V(0).Infof("Starting debug server for pprof at http://localhost%s/debug/pprof/", addr)
 		if err := http.ListenAndServe(addr, nil); err != nil && err != http.ErrServerClosed {
 			glog.Errorf("Failed to start debug server on %s: %v", addr, err)
 		}

weed/util/http/http_global_client_util.go

@@ -372,7 +372,7 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte,
 	var (
 		m int
 	)
-	buf := mem.Allocate(256 * 1024)
+	buf := mem.Allocate(64 * 1024)
 	defer mem.Free(buf)
 
 	for {