test: fix master client timeout causing test hangs

Use the main test context for KeepConnectedToMaster instead of creating a separate 60s context. The tests have 180s outer timeouts but the master client was disconnecting after 60s, causing subsequent commands to hang waiting for reconnection.
test: improve helper robustness
2026-10-13 18:15:50 +00:00 · 2025-12-10 23:40:48 -08:00 · 2025-12-10 22:42:28 -08:00 · 2025-12-10 22:41:35 -08:00 · 2025-12-10 22:27:12 -08:00 · 2025-12-10 21:33:11 -08:00
68 changed files with 5344 additions and 523 deletions
@@ -0,0 +1,92 @@
+name: "Metadata Subscribe Integration Tests"
+
+on:
+  push:
+    branches: [ master ]
+    paths:
+      - 'weed/filer/**'
+      - 'weed/pb/filer_pb/**'
+      - 'weed/util/log_buffer/**'
+      - 'weed/server/filer_grpc_server_sub_meta.go'
+      - 'weed/command/filer_backup.go'
+      - 'test/metadata_subscribe/**'
+      - '.github/workflows/metadata-subscribe-tests.yml'
+  pull_request:
+    branches: [ master ]
+    paths:
+      - 'weed/filer/**'
+      - 'weed/pb/filer_pb/**'
+      - 'weed/util/log_buffer/**'
+      - 'weed/server/filer_grpc_server_sub_meta.go'
+      - 'weed/command/filer_backup.go'
+      - 'test/metadata_subscribe/**'
+      - '.github/workflows/metadata-subscribe-tests.yml'
+
+concurrency:
+  group: ${{ github.head_ref }}/metadata-subscribe-tests
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+env:
+  GO_VERSION: '1.24'
+  TEST_TIMEOUT: '10m'
+
+jobs:
+  metadata-subscribe-integration:
+    name: Metadata Subscribe Integration Tests
+    runs-on: ubuntu-22.04
+    timeout-minutes: 20
+    
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v6
+      
+    - name: Set up Go ${{ env.GO_VERSION }}
+      uses: actions/setup-go@v6
+      with:
+        go-version: ${{ env.GO_VERSION }}
+        
+    - name: Build SeaweedFS
+      run: |
+        cd weed
+        go build -o weed .
+        chmod +x weed
+        ./weed version
+        
+    - name: Run Metadata Subscribe Integration Tests
+      run: |
+        cd test/metadata_subscribe
+        
+        echo "Running Metadata Subscribe integration tests..."
+        echo "============================================"
+        
+        # Run tests with verbose output
+        go test -v -timeout=${{ env.TEST_TIMEOUT }} ./...
+        
+        echo "============================================"
+        echo "Metadata Subscribe integration tests completed"
+        
+    - name: Archive logs on failure
+      if: failure()
+      uses: actions/upload-artifact@v5
+      with:
+        name: metadata-subscribe-test-logs
+        path: |
+          /tmp/seaweedfs_*
+        retention-days: 7
+        
+    - name: Test Summary
+      if: always()
+      run: |
+        echo "## Metadata Subscribe Integration Test Summary" >> $GITHUB_STEP_SUMMARY
+        echo "" >> $GITHUB_STEP_SUMMARY
+        echo "### Test Coverage" >> $GITHUB_STEP_SUMMARY
+        echo "- **Basic Subscription**: Subscribe to metadata changes and receive events" >> $GITHUB_STEP_SUMMARY
+        echo "- **Single-Filer No Stall**: Regression test for issue #4977" >> $GITHUB_STEP_SUMMARY
+        echo "- **Resume from Disk**: Verify subscription can resume from persisted logs" >> $GITHUB_STEP_SUMMARY
+        echo "" >> $GITHUB_STEP_SUMMARY
+        echo "### Related Issues" >> $GITHUB_STEP_SUMMARY
+        echo "- [#4977](https://github.com/seaweedfs/seaweedfs/issues/4977): filer.backup synchronisation stall" >> $GITHUB_STEP_SUMMARY
+
@@ -77,6 +77,7 @@ LABEL author="Chris Lu"
 RUN apt-get update && \
    apt-get install -y --no-install-recommends \
    ca-certificates \
+    curl \
    fuse \
    wget && \
    rm -rf /var/lib/apt/lists/*
@@ -27,7 +27,7 @@ COPY --from=builder /go/src/github.com/seaweedfs/seaweedfs/docker/entrypoint.sh
 # To disable: docker run -e GODEBUG=fips140=off ...

 # Install dependencies and create non-root user
-RUN apk add --no-cache fuse su-exec && \
+RUN apk add --no-cache fuse curl su-exec && \
    addgroup -g 1000 seaweed && \
    adduser -D -u 1000 -G seaweed seaweed

@@ -34,7 +34,7 @@ COPY --from=builder /go/src/github.com/seaweedfs/seaweedfs/docker/filer_rocksdb.
 COPY --from=builder /go/src/github.com/seaweedfs/seaweedfs/docker/entrypoint.sh /entrypoint.sh

 # Install dependencies and create non-root user
-RUN apk add --no-cache fuse snappy gflags su-exec && \
+RUN apk add --no-cache fuse snappy gflags curl su-exec && \
    addgroup -g 1000 seaweed && \
    adduser -D -u 1000 -G seaweed seaweed

@@ -17,7 +17,7 @@ COPY --from=builder /go/src/github.com/seaweedfs/seaweedfs/docker/filer_rocksdb.
 COPY --from=builder /go/src/github.com/seaweedfs/seaweedfs/docker/entrypoint.sh /entrypoint.sh

 # Install dependencies and create non-root user
-RUN apk add --no-cache fuse snappy gflags tmux su-exec && \
+RUN apk add --no-cache fuse snappy gflags curl tmux su-exec && \
    addgroup -g 1000 seaweed && \
    adduser -D -u 1000 -G seaweed seaweed

@@ -123,7 +123,7 @@ require (
 	github.com/Jille/raft-grpc-transport v1.6.1
 	github.com/ThreeDotsLabs/watermill v1.5.1
 	github.com/a-h/templ v0.3.943
-	github.com/apple/foundationdb/bindings/go v0.0.0-20240515141816-262c6fe778ad
+	github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3
 	github.com/arangodb/go-driver v1.6.9
 	github.com/armon/go-metrics v0.4.1
 	github.com/aws/aws-sdk-go-v2 v1.40.1
@@ -653,6 +653,8 @@ github.com/apache/arrow/go/v10 v10.0.1/go.mod h1:YvhnlEePVnBS4+0z3fhPfUy7W1Ikj0I
 github.com/apache/thrift v0.16.0/go.mod h1:PHK3hniurgQaNMZYaCLEqXKsYK8upmhPbmdP2FXSqgU=
 github.com/apple/foundationdb/bindings/go v0.0.0-20240515141816-262c6fe778ad h1:fQBkhYv86zyW95PWhzBlkgz3NoY1ue0L+8oYBaoCMbg=
 github.com/apple/foundationdb/bindings/go v0.0.0-20240515141816-262c6fe778ad/go.mod h1:OMVSB21p9+xQUIqlGizHPZfjK+SHws1ht+ZytVDoz9U=
+github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3 h1:WZaTKNHCfcw7fWSR6/RKnCldVzvYZC+Y20Su4lffEIg=
+github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3/go.mod h1:OMVSB21p9+xQUIqlGizHPZfjK+SHws1ht+ZytVDoz9U=
 github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc h1:LoL75er+LKDHDUfU5tRvFwxH0LjPpZN8OoG8Ll+liGU=
 github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc/go.mod h1:w648aMHEgFYS6xb0KVMMtZ2uMeemhiKCuD2vj6gY52A=
 github.com/arangodb/go-driver v1.6.9 h1:zckB+xuA16NmHUuYOX7INCJTIyIkoBQjAGqNpiyf2HQ=
@@ -145,6 +145,191 @@ stringData:
  seaweedfs_s3_config: '{"identities":[{"name":"anvAdmin","credentials":[{"accessKey":"snu8yoP6QAlY0ne4","secretKey":"PNzBcmeLNEdR0oviwm04NQAicOrDH1Km"}],"actions":["Admin","Read","Write"]},{"name":"anvReadOnly","credentials":[{"accessKey":"SCigFee6c5lbi04A","secretKey":"kgFhbT38R8WUYVtiFQ1OiSVOrYr3NKku"}],"actions":["Read"]}]}'
 ```

+## Admin Component
+
+The admin component provides a modern web-based administration interface for managing SeaweedFS clusters. It includes:
+
+- **Dashboard**: Real-time cluster status and metrics
+- **Volume Management**: Monitor volume servers, capacity, and health
+- **File Browser**: Browse and manage files in the filer
+- **Maintenance Operations**: Trigger maintenance tasks via workers
+- **Object Store Management**: Create and manage buckets with web interface
+
+### Enabling Admin
+
+To enable the admin interface, add the following to your values.yaml:
+
+```yaml
+admin:
+  enabled: true
+  port: 23646
+  grpcPort: 33646  # For worker connections
+  adminUser: "admin"
+  adminPassword: "your-secure-password"  # Leave empty to disable auth
+  
+  # Optional: persist admin data
+  data:
+    type: "persistentVolumeClaim"
+    size: "10Gi"
+    storageClass: "your-storage-class"
+  
+  # Optional: enable ingress
+  ingress:
+    enabled: true
+    host: "admin.seaweedfs.local"
+    className: "nginx"
+```
+
+The admin interface will be available at `http://<admin-service>:23646` (or via ingress). Workers connect to the admin server via gRPC on port `33646`.
+
+### Admin Authentication
+
+If `adminPassword` is set, the admin interface requires authentication:
+- Username: Value of `adminUser` (default: `admin`)
+- Password: Value of `adminPassword`
+
+If `adminPassword` is empty or not set, the admin interface runs without authentication (not recommended for production).
+
+### Admin Data Persistence
+
+The admin component can store configuration and maintenance data. You can configure storage in several ways:
+
+- **emptyDir** (default): Data is lost when pod restarts
+- **persistentVolumeClaim**: Data persists across pod restarts
+- **hostPath**: Data stored on the host filesystem
+- **existingClaim**: Use an existing PVC
+
+## Worker Component
+
+Workers are maintenance agents that execute cluster maintenance tasks such as vacuum, volume balancing, and erasure coding. Workers connect to the admin server via gRPC and receive task assignments.
+
+### Enabling Workers
+
+To enable workers, add the following to your values.yaml:
+
+```yaml
+worker:
+  enabled: true
+  replicas: 2  # Scale based on workload
+  capabilities: "vacuum,balance,erasure_coding"  # Tasks this worker can handle
+  maxConcurrent: 3  # Maximum concurrent tasks per worker
+  
+  # Working directory for task execution
+  # Default: "/tmp/seaweedfs-worker"
+  # Note: /tmp is ephemeral - use persistent storage (hostPath/existingClaim) for long-running tasks
+  workingDir: "/tmp/seaweedfs-worker"
+  
+  # Optional: configure admin server address
+  # If not specified, auto-discovers from admin service in the same namespace by looking for
+  # a service named "<release-name>-admin" (e.g., "seaweedfs-admin").
+  # Auto-discovery only works if the admin is in the same namespace and same Helm release.
+  # For cross-namespace or separate release scenarios, explicitly set this value.
+  # Example: If main SeaweedFS is deployed in "production" namespace:
+  #   adminServer: "seaweedfs-admin.production.svc:33646"
+  adminServer: ""
+  
+  # Workers need storage for task execution
+  # Note: Workers use a Deployment, which does not support `volumeClaimTemplates` 
+  # for dynamic PVC creation per pod. To use persistent storage, you must 
+  # pre-provision a PersistentVolumeClaim and use `type: "existingClaim"`.
+  data:
+    type: "emptyDir"  # Options: "emptyDir", "hostPath", or "existingClaim"
+    hostPathPrefix: /storage  # For hostPath
+    # claimName: "worker-pvc"  # For existingClaim with pre-provisioned PVC
+  
+  # Resource limits for worker pods
+  resources:
+    requests:
+      cpu: "500m"
+      memory: "512Mi"
+    limits:
+      cpu: "2"
+      memory: "2Gi"
+```
+
+### Worker Capabilities
+
+Workers can be configured with different capabilities:
+- **vacuum**: Reclaim deleted file space
+- **balance**: Balance volumes across volume servers
+- **erasure_coding**: Handle erasure coding operations
+
+You can configure workers with all capabilities or create specialized worker pools with specific capabilities.
+
+### Worker Deployment Strategy
+
+For production deployments, consider:
+
+1. **Multiple Workers**: Deploy 2+ worker replicas for high availability
+2. **Resource Allocation**: Workers need sufficient CPU/memory for maintenance tasks
+3. **Storage**: Workers need temporary storage for vacuum and balance operations (size depends on volume size)
+4. **Specialized Workers**: Create separate worker deployments for different capabilities if needed
+
+Example specialized worker configuration:
+
+For specialized worker pools, deploy separate Helm releases with different capabilities:
+
+**values-worker-vacuum.yaml** (for vacuum operations):
+```yaml
+# Disable all other components, enable only workers
+master:
+  enabled: false
+volume:
+  enabled: false
+filer:
+  enabled: false
+s3:
+  enabled: false
+admin:
+  enabled: false
+
+worker:
+  enabled: true
+  replicas: 2
+  capabilities: "vacuum"
+  maxConcurrent: 2
+  # REQUIRED: Point to the admin service of your main SeaweedFS release
+  # Replace <namespace> with the namespace where your main seaweedfs is deployed
+  # Example: If deploying in namespace "production":
+  #   adminServer: "seaweedfs-admin.production.svc:33646"
+  adminServer: "seaweedfs-admin.<namespace>.svc:33646"
+```
+
+**values-worker-balance.yaml** (for balance operations):
+```yaml
+# Disable all other components, enable only workers
+master:
+  enabled: false
+volume:
+  enabled: false
+filer:
+  enabled: false
+s3:
+  enabled: false
+admin:
+  enabled: false
+
+worker:
+  enabled: true
+  replicas: 1
+  capabilities: "balance"
+  maxConcurrent: 1
+  # REQUIRED: Point to the admin service of your main SeaweedFS release
+  # Replace <namespace> with the namespace where your main seaweedfs is deployed
+  # Example: If deploying in namespace "production":
+  #   adminServer: "seaweedfs-admin.production.svc:33646"
+  adminServer: "seaweedfs-admin.<namespace>.svc:33646"
+```
+
+Deploy the specialized workers as separate releases:
+```bash
+# Deploy vacuum workers
+helm install seaweedfs-worker-vacuum seaweedfs/seaweedfs -f values-worker-vacuum.yaml
+
+# Deploy balance workers
+helm install seaweedfs-worker-balance seaweedfs/seaweedfs -f values-worker-balance.yaml
+```
+
 ## Enterprise

 For enterprise users, please visit [seaweedfs.com](https://seaweedfs.com) for the SeaweedFS Enterprise Edition, 
@@ -0,0 +1,52 @@
+{{- if and .Values.admin.enabled .Values.admin.ingress.enabled }}
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion }}
+apiVersion: networking.k8s.io/v1
+{{- else if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion }}
+apiVersion: networking.k8s.io/v1beta1
+{{- else }}
+apiVersion: extensions/v1beta1
+{{- end }}
+kind: Ingress
+metadata:
+  name: ingress-{{ template "seaweedfs.name" . }}-admin
+  namespace: {{ .Release.Namespace }}
+  annotations:
+  {{- if and (not (semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion)) .Values.admin.ingress.className }}
+    kubernetes.io/ingress.class: {{ .Values.admin.ingress.className }}
+  {{- end }}
+  {{- with .Values.admin.ingress.annotations }}
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: admin
+spec:
+  {{- if and (semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion) .Values.admin.ingress.className }}
+  ingressClassName: {{ .Values.admin.ingress.className | quote }}
+  {{- end }}
+  tls:
+    {{ .Values.admin.ingress.tls | default list | toYaml | nindent 6}}
+  rules:
+  - {{- if .Values.admin.ingress.host }}
+    host: {{ .Values.admin.ingress.host | quote }}
+    {{- end }}
+    http:
+      paths:
+      - path: {{ .Values.admin.ingress.path | quote }}
+        {{- if semverCompare ">=1.18-0" .Capabilities.KubeVersion.GitVersion }}
+        pathType: {{ .Values.admin.ingress.pathType | quote }}
+        {{- end }}
+        backend:
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion }}
+          service:
+            name: {{ template "seaweedfs.name" . }}-admin
+            port:
+              number: {{ .Values.admin.port }}
+{{- else }}
+          serviceName: {{ template "seaweedfs.name" . }}-admin
+          servicePort: {{ .Values.admin.port }}
+{{- end }}
+{{- end }}
@@ -0,0 +1,20 @@
+{{- if and .Values.admin.enabled .Values.admin.secret.adminPassword (not .Values.admin.secret.existingSecret) }}
+apiVersion: v1
+kind: Secret
+type: Opaque
+metadata:
+  name: {{ template "seaweedfs.name" . }}-admin-secret
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    "helm.sh/resource-policy": keep
+    "helm.sh/hook": "pre-install,pre-upgrade"
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: admin
+data:
+  adminUser: {{ .Values.admin.secret.adminUser | b64enc }}
+  adminPassword: {{ .Values.admin.secret.adminPassword | b64enc }}
+{{- end}}
@@ -0,0 +1,39 @@
+{{- if .Values.admin.enabled }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ template "seaweedfs.name" . }}-admin
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: admin
+{{- if .Values.admin.service.annotations }}
+  annotations:
+    {{- toYaml .Values.admin.service.annotations | nindent 4 }}
+{{- end }}
+spec:
+  type: {{ .Values.admin.service.type }}
+  ports:
+  - name: "http"
+    port: {{ .Values.admin.port }}
+    targetPort: {{ .Values.admin.port }}
+    protocol: TCP
+  - name: "grpc"
+    port: {{ .Values.admin.grpcPort }}
+    targetPort: {{ .Values.admin.grpcPort }}
+    protocol: TCP
+  {{- if .Values.admin.metricsPort }}
+  - name: "metrics"
+    port: {{ .Values.admin.metricsPort }}
+    targetPort: {{ .Values.admin.metricsPort }}
+    protocol: TCP
+  {{- end }}
+  selector:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: admin
+{{- end }}
+
@@ -0,0 +1,33 @@
+{{- if .Values.admin.enabled }}
+{{- if .Values.admin.metricsPort }}
+{{- if .Values.global.monitoring.enabled }}
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: {{ template "seaweedfs.name" . }}-admin
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: admin
+    {{- with .Values.global.monitoring.additionalLabels }}
+    {{- toYaml . | nindent 4 }}
+    {{- end }}
+{{- with .Values.admin.serviceMonitor.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+{{- end }}
+spec:
+  endpoints:
+    - interval: 30s
+      port: metrics
+      scrapeTimeout: 5s
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+      app.kubernetes.io/component: admin
+{{- end }}
+{{- end }}
+{{- end }}
@@ -0,0 +1,339 @@
+{{- if .Values.admin.enabled }}
+{{- if and (not .Values.admin.masters) (not .Values.global.masterServer) (not .Values.master.enabled) }}
+{{- fail "admin.masters or global.masterServer must be set if master.enabled is false" -}}
+{{- end }}
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: {{ template "seaweedfs.name" . }}-admin
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: admin
+{{- if .Values.admin.annotations }}
+  annotations:
+    {{- toYaml .Values.admin.annotations | nindent 4 }}
+{{- end }}
+spec:
+  serviceName: {{ template "seaweedfs.name" . }}-admin
+  podManagementPolicy: {{ .Values.admin.podManagementPolicy }}
+  replicas: {{ .Values.admin.replicas }}
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+      app.kubernetes.io/instance: {{ .Release.Name }}
+      app.kubernetes.io/component: admin
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+        helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+        app.kubernetes.io/instance: {{ .Release.Name }}
+        app.kubernetes.io/component: admin
+      {{ with .Values.podLabels }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.admin.podLabels }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+      annotations:
+      {{ with .Values.podAnnotations }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.admin.podAnnotations }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+    spec:
+      restartPolicy: {{ default .Values.global.restartPolicy .Values.admin.restartPolicy }}
+      {{- if .Values.admin.affinity }}
+      affinity:
+        {{ tpl .Values.admin.affinity . | nindent 8 | trim }}
+      {{- end }}
+      {{- if .Values.admin.topologySpreadConstraints }}
+      topologySpreadConstraints:
+        {{ tpl .Values.admin.topologySpreadConstraints . | nindent 8 | trim }}
+      {{- end }}
+      {{- if .Values.admin.tolerations }}
+      tolerations:
+        {{ tpl .Values.admin.tolerations . | nindent 8 | trim }}
+      {{- end }}
+      {{- include "seaweedfs.imagePullSecrets" . | nindent 6 }}
+      terminationGracePeriodSeconds: 30
+      {{- if .Values.admin.priorityClassName }}
+      priorityClassName: {{ .Values.admin.priorityClassName | quote }}
+      {{- end }}
+      enableServiceLinks: false
+      {{- if .Values.admin.serviceAccountName }}
+      serviceAccountName: {{ .Values.admin.serviceAccountName | quote }}
+      {{- end }}
+      {{- if .Values.admin.initContainers }}
+      initContainers:
+        {{ tpl .Values.admin.initContainers . | nindent 8 | trim }}
+      {{- end }}
+      {{- if .Values.admin.podSecurityContext.enabled }}
+      securityContext: {{- omit .Values.admin.podSecurityContext "enabled" | toYaml | nindent 8 }}
+      {{- end }}
+      containers:
+        - name: seaweedfs
+          image: {{ template "admin.image" . }}
+          imagePullPolicy: {{ default "IfNotPresent" .Values.global.imagePullPolicy }}
+          {{- $adminAuthEnabled := or .Values.admin.secret.existingSecret .Values.admin.secret.adminPassword }}
+          {{- if and .Values.admin.secret.existingSecret (not .Values.admin.secret.userKey) -}}
+          {{-   fail "admin.secret.userKey must be set when admin.secret.existingSecret is provided" -}}
+          {{- end -}}
+          {{- if and .Values.admin.secret.existingSecret (not .Values.admin.secret.pwKey) -}}
+          {{-   fail "admin.secret.pwKey must be set when admin.secret.existingSecret is provided" -}}
+          {{- end -}}
+          {{- $adminSecretName := .Values.admin.secret.existingSecret | default (printf "%s-admin-secret" (include "seaweedfs.name" .)) }}
+          env:
+            {{- if $adminAuthEnabled }}
+            - name: SEAWEEDFS_ADMIN_USER
+              valueFrom:
+                secretKeyRef:
+                  name: {{ $adminSecretName }}
+                  key: {{ if .Values.admin.secret.existingSecret }}{{ .Values.admin.secret.userKey }}{{ else }}adminUser{{ end }}
+            - name: SEAWEEDFS_ADMIN_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: {{ $adminSecretName }}
+                  key: {{ if .Values.admin.secret.existingSecret }}{{ .Values.admin.secret.pwKey }}{{ else }}adminPassword{{ end }}
+            {{- end }}
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+            - name: SEAWEEDFS_FULLNAME
+              value: "{{ template "seaweedfs.name" . }}"
+            {{- if .Values.admin.extraEnvironmentVars }}
+            {{- range $key, $value := .Values.admin.extraEnvironmentVars }}
+            - name: {{ $key }}
+            {{- if kindIs "string" $value }}
+              value: {{ $value | quote }}
+            {{- else }}
+              valueFrom:
+                {{ toYaml $value | nindent 16 | trim }}
+            {{- end -}}
+            {{- end }}
+            {{- end }}
+            {{- if .Values.global.extraEnvironmentVars }}
+            {{- range $key, $value := .Values.global.extraEnvironmentVars }}
+            - name: {{ $key }}
+            {{- if kindIs "string" $value }}
+              value: {{ $value | quote }}
+            {{- else }}
+              valueFrom:
+                {{ toYaml $value | nindent 16 | trim }}
+            {{- end -}}
+            {{- end }}
+            {{- end }}
+          command:
+            - "/bin/sh"
+            - "-ec"
+            - |
+              exec /usr/bin/weed \
+              {{- if or (eq .Values.admin.logs.type "hostPath") (eq .Values.admin.logs.type "persistentVolumeClaim") (eq .Values.admin.logs.type "emptyDir") (eq .Values.admin.logs.type "existingClaim") }}
+              -logdir=/logs \
+              {{- else }}
+              -logtostderr=true \
+              {{- end }}
+              {{- if .Values.admin.loggingOverrideLevel }}
+              -v={{ .Values.admin.loggingOverrideLevel }} \
+              {{- else }}
+              -v={{ .Values.global.loggingLevel }} \
+              {{- end }}
+              admin \
+              -port={{ .Values.admin.port }} \
+              -port.grpc={{ .Values.admin.grpcPort }} \
+              {{- if or (eq .Values.admin.data.type "hostPath") (eq .Values.admin.data.type "persistentVolumeClaim") (eq .Values.admin.data.type "emptyDir") (eq .Values.admin.data.type "existingClaim") }}
+              -dataDir=/data \
+              {{- else if .Values.admin.dataDir }}
+              -dataDir={{ .Values.admin.dataDir }} \
+              {{- end }}
+              {{- if $adminAuthEnabled }}
+              -adminUser="${SEAWEEDFS_ADMIN_USER}" \
+              -adminPassword="${SEAWEEDFS_ADMIN_PASSWORD}" \
+              {{- end }}
+              {{- if .Values.admin.masters }}
+              -masters={{ .Values.admin.masters }}{{- if .Values.admin.extraArgs }} \{{ end }}
+              {{- else if .Values.global.masterServer }}
+              -masters={{ .Values.global.masterServer }}{{- if .Values.admin.extraArgs }} \{{ end }}
+              {{- else }}
+              -masters={{ range $index := until (.Values.master.replicas | int) }}${SEAWEEDFS_FULLNAME}-master-{{ $index }}.${SEAWEEDFS_FULLNAME}-master.{{ $.Release.Namespace }}:{{ $.Values.master.port }}{{ if lt $index (sub ($.Values.master.replicas | int) 1) }},{{ end }}{{ end }}{{- if .Values.admin.extraArgs }} \{{ end }}
+              {{- end }}
+              {{- range $index, $arg := .Values.admin.extraArgs }}
+              {{ $arg }}{{- if lt $index (sub (len $.Values.admin.extraArgs) 1) }} \{{ end }}
+              {{- end }}
+          volumeMounts:
+            {{- if or (eq .Values.admin.data.type "hostPath") (eq .Values.admin.data.type "persistentVolumeClaim") (eq .Values.admin.data.type "emptyDir") (eq .Values.admin.data.type "existingClaim") }}
+            - name: admin-data
+              mountPath: /data
+            {{- end }}
+            {{- if or (eq .Values.admin.logs.type "hostPath") (eq .Values.admin.logs.type "persistentVolumeClaim") (eq .Values.admin.logs.type "emptyDir") (eq .Values.admin.logs.type "existingClaim") }}
+            - name: admin-logs
+              mountPath: /logs
+            {{- end }}
+            {{- if .Values.global.enableSecurity }}
+            - name: security-config
+              readOnly: true
+              mountPath: /etc/seaweedfs/security.toml
+              subPath: security.toml
+            - name: ca-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/ca/
+            - name: master-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/master/
+            - name: volume-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/volume/
+            - name: filer-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/filer/
+            - name: client-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/client/
+            {{- end }}
+            {{ tpl .Values.admin.extraVolumeMounts . | nindent 12 | trim }}
+          ports:
+            - containerPort: {{ .Values.admin.port }}
+              name: http
+            - containerPort: {{ .Values.admin.grpcPort }}
+              name: grpc
+            {{- if .Values.admin.metricsPort }}
+            - containerPort: {{ .Values.admin.metricsPort }}
+              name: metrics
+            {{- end }}
+          {{- if .Values.admin.readinessProbe.enabled }}
+          readinessProbe:
+            httpGet:
+              path: {{ .Values.admin.readinessProbe.httpGet.path }}
+              port: http
+              scheme: {{ .Values.admin.readinessProbe.httpGet.scheme }}
+            initialDelaySeconds: {{ .Values.admin.readinessProbe.initialDelaySeconds }}
+            periodSeconds: {{ .Values.admin.readinessProbe.periodSeconds }}
+            successThreshold: {{ .Values.admin.readinessProbe.successThreshold }}
+            failureThreshold: {{ .Values.admin.readinessProbe.failureThreshold }}
+            timeoutSeconds: {{ .Values.admin.readinessProbe.timeoutSeconds }}
+          {{- end }}
+          {{- if .Values.admin.livenessProbe.enabled }}
+          livenessProbe:
+            httpGet:
+              path: {{ .Values.admin.livenessProbe.httpGet.path }}
+              port: http
+              scheme: {{ .Values.admin.livenessProbe.httpGet.scheme }}
+            initialDelaySeconds: {{ .Values.admin.livenessProbe.initialDelaySeconds }}
+            periodSeconds: {{ .Values.admin.livenessProbe.periodSeconds }}
+            successThreshold: {{ .Values.admin.livenessProbe.successThreshold }}
+            failureThreshold: {{ .Values.admin.livenessProbe.failureThreshold }}
+            timeoutSeconds: {{ .Values.admin.livenessProbe.timeoutSeconds }}
+          {{- end }}
+          {{- with .Values.admin.resources }}
+          resources:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- if .Values.admin.containerSecurityContext.enabled }}
+          securityContext: {{- omit .Values.admin.containerSecurityContext "enabled" | toYaml | nindent 12 }}
+          {{- end }}
+      {{- if .Values.admin.sidecars }}
+      {{- include "common.tplvalues.render" (dict "value" .Values.admin.sidecars "context" $) | nindent 8 }}
+      {{- end }}
+      volumes:
+        {{- if eq .Values.admin.data.type "hostPath" }}
+        - name: admin-data
+          hostPath:
+            path: {{ .Values.admin.data.hostPathPrefix }}/seaweedfs-admin-data
+            type: DirectoryOrCreate
+        {{- end }}
+        {{- if eq .Values.admin.data.type "emptyDir" }}
+        - name: admin-data
+          emptyDir: {}
+        {{- end }}
+        {{- if eq .Values.admin.data.type "existingClaim" }}
+        - name: admin-data
+          persistentVolumeClaim:
+            claimName: {{ .Values.admin.data.claimName }}
+        {{- end }}
+        {{- if eq .Values.admin.logs.type "hostPath" }}
+        - name: admin-logs
+          hostPath:
+            path: {{ .Values.admin.logs.hostPathPrefix }}/logs/seaweedfs/admin
+            type: DirectoryOrCreate
+        {{- end }}
+        {{- if eq .Values.admin.logs.type "emptyDir" }}
+        - name: admin-logs
+          emptyDir: {}
+        {{- end }}
+        {{- if eq .Values.admin.logs.type "existingClaim" }}
+        - name: admin-logs
+          persistentVolumeClaim:
+            claimName: {{ .Values.admin.logs.claimName }}
+        {{- end }}
+        {{- if .Values.global.enableSecurity }}
+        - name: security-config
+          configMap:
+            name: {{ template "seaweedfs.name" . }}-security-config
+        - name: ca-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-ca-cert
+        - name: master-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-master-cert
+        - name: volume-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-volume-cert
+        - name: filer-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-filer-cert
+        - name: client-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-client-cert
+        {{- end }}
+        {{ tpl .Values.admin.extraVolumes . | indent 8 | trim }}
+      {{- if .Values.admin.nodeSelector }}
+      nodeSelector:
+        {{ tpl .Values.admin.nodeSelector . | indent 8 | trim }}
+      {{- end }}
+  {{- $pvc_exists := include "admin.pvc_exists" . -}}
+  {{- if $pvc_exists }}
+  volumeClaimTemplates:
+    {{- if eq .Values.admin.data.type "persistentVolumeClaim" }}
+    - metadata:
+        name: admin-data
+        {{- with .Values.admin.data.annotations }}
+        annotations:
+          {{- toYaml . | nindent 10 }}
+        {{- end }}
+      spec:
+        accessModes: [ "ReadWriteOnce" ]
+        storageClassName: {{ .Values.admin.data.storageClass }}
+        resources:
+          requests:
+            storage: {{ .Values.admin.data.size }}
+    {{- end }}
+    {{- if eq .Values.admin.logs.type "persistentVolumeClaim" }}
+    - metadata:
+        name: admin-logs
+        {{- with .Values.admin.logs.annotations }}
+        annotations:
+          {{- toYaml . | nindent 10 }}
+        {{- end }}
+      spec:
+        accessModes: [ "ReadWriteOnce" ]
+        storageClassName: {{ .Values.admin.logs.storageClass }}
+        resources:
+          requests:
+            storage: {{ .Values.admin.logs.size }}
+    {{- end }}
+  {{- end }}
+{{- end }}
@@ -83,6 +83,26 @@ Inject extra environment vars in the format key:value, if populated
 {{- end -}}
 {{- end -}}

+{{/* Return the proper admin image */}}
+{{- define "admin.image" -}}
+{{- if .Values.admin.imageOverride -}}
+{{- $imageOverride := .Values.admin.imageOverride -}}
+{{- printf "%s" $imageOverride -}}
+{{- else -}}
+{{- include "common.image" . }}
+{{- end -}}
+{{- end -}}
+
+{{/* Return the proper worker image */}}
+{{- define "worker.image" -}}
+{{- if .Values.worker.imageOverride -}}
+{{- $imageOverride := .Values.worker.imageOverride -}}
+{{- printf "%s" $imageOverride -}}
+{{- else -}}
+{{- include "common.image" . }}
+{{- end -}}
+{{- end -}}
+
 {{/* Return the proper volume image */}}
 {{- define "volume.image" -}}
 {{- if .Values.volume.imageOverride -}}
@@ -136,6 +156,15 @@ Inject extra environment vars in the format key:value, if populated
 {{- end -}}
 {{- end -}}

+{{/* check if any Admin PVC exists */}}
+{{- define "admin.pvc_exists" -}}
+{{- if or (eq .Values.admin.data.type "persistentVolumeClaim") (eq .Values.admin.logs.type "persistentVolumeClaim") -}}
+{{- printf "true" -}}
+{{- else -}}
+{{- printf "" -}}
+{{- end -}}
+{{- end -}}
+
 {{/* check if any InitContainers exist for Volumes */}}
 {{- define "volume.initContainers_exists" -}}
 {{- if or (not (empty .Values.volume.idx )) (not (empty .Values.volume.initContainers )) -}}
@@ -0,0 +1,271 @@
+{{- if .Values.worker.enabled }}
+{{- if and (not .Values.worker.adminServer) (not .Values.admin.enabled) }}
+{{- fail "worker.adminServer must be set if admin.enabled is false within the same release" -}}
+{{- end }}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ template "seaweedfs.name" . }}-worker
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: worker
+{{- if .Values.worker.annotations }}
+  annotations:
+    {{- toYaml .Values.worker.annotations | nindent 4 }}
+{{- end }}
+spec:
+  replicas: {{ .Values.worker.replicas }}
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+      app.kubernetes.io/instance: {{ .Release.Name }}
+      app.kubernetes.io/component: worker
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+        helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+        app.kubernetes.io/instance: {{ .Release.Name }}
+        app.kubernetes.io/component: worker
+      {{ with .Values.podLabels }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.worker.podLabels }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+      annotations:
+      {{ with .Values.podAnnotations }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.worker.podAnnotations }}
+      {{- toYaml . | nindent 8 }}
+      {{- end }}
+    spec:
+      restartPolicy: {{ default .Values.global.restartPolicy .Values.worker.restartPolicy }}
+      {{- if .Values.worker.affinity }}
+      affinity:
+        {{ tpl .Values.worker.affinity . | nindent 8 | trim }}
+      {{- end }}
+      {{- if .Values.worker.topologySpreadConstraints }}
+      topologySpreadConstraints:
+        {{ tpl .Values.worker.topologySpreadConstraints . | nindent 8 | trim }}
+      {{- end }}
+      {{- if .Values.worker.tolerations }}
+      tolerations:
+        {{ tpl .Values.worker.tolerations . | nindent 8 | trim }}
+      {{- end }}
+      {{- include "seaweedfs.imagePullSecrets" . | nindent 6 }}
+      terminationGracePeriodSeconds: 60
+      {{- if .Values.worker.priorityClassName }}
+      priorityClassName: {{ .Values.worker.priorityClassName | quote }}
+      {{- end }}
+      enableServiceLinks: false
+      {{- if .Values.worker.serviceAccountName }}
+      serviceAccountName: {{ .Values.worker.serviceAccountName | quote }}
+      {{- end }}
+      {{- if .Values.worker.initContainers }}
+      initContainers:
+        {{ tpl .Values.worker.initContainers . | nindent 8 | trim }}
+      {{- end }}
+      {{- if .Values.worker.podSecurityContext.enabled }}
+      securityContext: {{- omit .Values.worker.podSecurityContext "enabled" | toYaml | nindent 8 }}
+      {{- end }}
+      containers:
+        - name: seaweedfs
+          image: {{ template "worker.image" . }}
+          imagePullPolicy: {{ default "IfNotPresent" .Values.global.imagePullPolicy }}
+          env:
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+            - name: SEAWEEDFS_FULLNAME
+              value: "{{ template "seaweedfs.name" . }}"
+            {{- if .Values.worker.extraEnvironmentVars }}
+            {{- range $key, $value := .Values.worker.extraEnvironmentVars }}
+            - name: {{ $key }}
+            {{- if kindIs "string" $value }}
+              value: {{ $value | quote }}
+            {{- else }}
+              valueFrom:
+                {{ toYaml $value | nindent 16 | trim }}
+            {{- end -}}
+            {{- end }}
+            {{- end }}
+            {{- if .Values.global.extraEnvironmentVars }}
+            {{- range $key, $value := .Values.global.extraEnvironmentVars }}
+            - name: {{ $key }}
+            {{- if kindIs "string" $value }}
+              value: {{ $value | quote }}
+            {{- else }}
+              valueFrom:
+                {{ toYaml $value | nindent 16 | trim }}
+            {{- end -}}
+            {{- end }}
+            {{- end }}
+          command:
+            - "/bin/sh"
+            - "-ec"
+            - |
+              exec /usr/bin/weed \
+              {{- if or (eq .Values.worker.logs.type "hostPath") (eq .Values.worker.logs.type "emptyDir") (eq .Values.worker.logs.type "existingClaim") }}
+              -logdir=/logs \
+              {{- else }}
+              -logtostderr=true \
+              {{- end }}
+              {{- if .Values.worker.loggingOverrideLevel }}
+              -v={{ .Values.worker.loggingOverrideLevel }} \
+              {{- else }}
+              -v={{ .Values.global.loggingLevel }} \
+              {{- end }}
+              worker \
+              {{- if .Values.worker.adminServer }}
+              -admin={{ .Values.worker.adminServer }} \
+              {{- else }}
+              -admin={{ template "seaweedfs.name" . }}-admin.{{ .Release.Namespace }}:{{ .Values.admin.grpcPort }} \
+              {{- end }}
+              -capabilities={{ .Values.worker.capabilities }} \
+              -maxConcurrent={{ .Values.worker.maxConcurrent }} \
+              -workingDir={{ .Values.worker.workingDir }}{{- if .Values.worker.extraArgs }} \{{ end }}
+              {{- range $index, $arg := .Values.worker.extraArgs }}
+              {{ $arg }}{{- if lt $index (sub (len $.Values.worker.extraArgs) 1) }} \{{ end }}
+              {{- end }}
+          volumeMounts:
+            {{- if or (eq .Values.worker.data.type "hostPath") (eq .Values.worker.data.type "emptyDir") (eq .Values.worker.data.type "existingClaim") }}
+            - name: worker-data
+              mountPath: {{ .Values.worker.workingDir }}
+            {{- end }}
+            {{- if or (eq .Values.worker.logs.type "hostPath") (eq .Values.worker.logs.type "emptyDir") (eq .Values.worker.logs.type "existingClaim") }}
+            - name: worker-logs
+              mountPath: /logs
+            {{- end }}
+            {{- if .Values.global.enableSecurity }}
+            - name: security-config
+              readOnly: true
+              mountPath: /etc/seaweedfs/security.toml
+              subPath: security.toml
+            - name: ca-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/ca/
+            - name: master-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/master/
+            - name: volume-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/volume/
+            - name: filer-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/filer/
+            - name: client-cert
+              readOnly: true
+              mountPath: /usr/local/share/ca-certificates/client/
+            {{- end }}
+            {{ tpl .Values.worker.extraVolumeMounts . | nindent 12 | trim }}
+          ports:
+            {{- if .Values.worker.metricsPort }}
+            - containerPort: {{ .Values.worker.metricsPort }}
+              name: metrics
+            {{- end }}
+          {{- with .Values.worker.resources }}
+          resources:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- if .Values.worker.livenessProbe.enabled }}
+          livenessProbe:
+            {{- with .Values.worker.livenessProbe.tcpSocket }}
+            tcpSocket:
+              port: {{ .port }}
+            {{- end }}
+            initialDelaySeconds: {{ .Values.worker.livenessProbe.initialDelaySeconds }}
+            periodSeconds: {{ .Values.worker.livenessProbe.periodSeconds }}
+            successThreshold: {{ .Values.worker.livenessProbe.successThreshold }}
+            failureThreshold: {{ .Values.worker.livenessProbe.failureThreshold }}
+            timeoutSeconds: {{ .Values.worker.livenessProbe.timeoutSeconds }}
+          {{- end }}
+          {{- if .Values.worker.readinessProbe.enabled }}
+          readinessProbe:
+            {{- with .Values.worker.readinessProbe.tcpSocket }}
+            tcpSocket:
+              port: {{ .port }}
+            {{- end }}
+            initialDelaySeconds: {{ .Values.worker.readinessProbe.initialDelaySeconds }}
+            periodSeconds: {{ .Values.worker.readinessProbe.periodSeconds }}
+            successThreshold: {{ .Values.worker.readinessProbe.successThreshold }}
+            failureThreshold: {{ .Values.worker.readinessProbe.failureThreshold }}
+            timeoutSeconds: {{ .Values.worker.readinessProbe.timeoutSeconds }}
+          {{- end }}
+          {{- if .Values.worker.containerSecurityContext.enabled }}
+          securityContext: {{- omit .Values.worker.containerSecurityContext "enabled" | toYaml | nindent 12 }}
+          {{- end }}
+      {{- if .Values.worker.sidecars }}
+      {{- include "common.tplvalues.render" (dict "value" .Values.worker.sidecars "context" $) | nindent 8 }}
+      {{- end }}
+      volumes:
+        {{- if eq .Values.worker.data.type "hostPath" }}
+        - name: worker-data
+          hostPath:
+            path: {{ .Values.worker.data.hostPathPrefix }}/seaweedfs-worker-data
+            type: DirectoryOrCreate
+        {{- end }}
+        {{- if eq .Values.worker.data.type "emptyDir" }}
+        - name: worker-data
+          emptyDir: {}
+        {{- end }}
+        {{- if eq .Values.worker.data.type "existingClaim" }}
+        - name: worker-data
+          persistentVolumeClaim:
+            claimName: {{ .Values.worker.data.claimName }}
+        {{- end }}
+        {{- if eq .Values.worker.logs.type "hostPath" }}
+        - name: worker-logs
+          hostPath:
+            path: {{ .Values.worker.logs.hostPathPrefix }}/logs/seaweedfs/worker
+            type: DirectoryOrCreate
+        {{- end }}
+        {{- if eq .Values.worker.logs.type "emptyDir" }}
+        - name: worker-logs
+          emptyDir: {}
+        {{- end }}
+        {{- if eq .Values.worker.logs.type "existingClaim" }}
+        - name: worker-logs
+          persistentVolumeClaim:
+            claimName: {{ .Values.worker.logs.claimName }}
+        {{- end }}
+        {{- if .Values.global.enableSecurity }}
+        - name: security-config
+          configMap:
+            name: {{ template "seaweedfs.name" . }}-security-config
+        - name: ca-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-ca-cert
+        - name: master-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-master-cert
+        - name: volume-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-volume-cert
+        - name: filer-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-filer-cert
+        - name: client-cert
+          secret:
+            secretName: {{ template "seaweedfs.name" . }}-client-cert
+        {{- end }}
+        {{ tpl .Values.worker.extraVolumes . | indent 8 | trim }}
+      {{- if .Values.worker.nodeSelector }}
+      nodeSelector:
+        {{ tpl .Values.worker.nodeSelector . | indent 8 | trim }}
+      {{- end }}
+{{- end }}
@@ -0,0 +1,26 @@
+{{- if .Values.worker.enabled }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ template "seaweedfs.name" . }}-worker
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: worker
+spec:
+  clusterIP: None  # Headless service
+  {{- if .Values.worker.metricsPort }}
+  ports:
+  - name: "metrics"
+    port: {{ .Values.worker.metricsPort }}
+    targetPort: {{ .Values.worker.metricsPort }}
+    protocol: TCP
+  {{- end }}
+  selector:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: worker
+{{- end }}
@@ -0,0 +1,33 @@
+{{- if .Values.worker.enabled }}
+{{- if .Values.worker.metricsPort }}
+{{- if .Values.global.monitoring.enabled }}
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: {{ template "seaweedfs.name" . }}-worker
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+    helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: worker
+    {{- with .Values.global.monitoring.additionalLabels }}
+    {{- toYaml . | nindent 4 }}
+    {{- end }}
+{{- with .Values.worker.serviceMonitor.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+{{- end }}
+spec:
+  endpoints:
+    - interval: 30s
+      port: metrics
+      scrapeTimeout: 5s
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+      app.kubernetes.io/component: worker
+{{- end }}
+{{- end }}
+{{- end }}
@@ -1088,6 +1088,245 @@ sftp:
    failureThreshold: 100
    timeoutSeconds: 10

+admin:
+  enabled: false
+  imageOverride: null
+  restartPolicy: null
+  replicas: 1
+  port: 23646  # Default admin port
+  grpcPort: 33646  # Default gRPC port for worker connections
+  metricsPort: 9327
+  loggingOverrideLevel: null
+
+  # Admin authentication
+  secret:
+    # Name of an existing secret containing admin credentials. If set, adminUser and adminPassword below are ignored.
+    existingSecret: ""
+    # Key in the existing secret for the admin username. Required if existingSecret is set.
+    userKey: ""
+    # Key in the existing secret for the admin password. Required if existingSecret is set.
+    pwKey: ""
+    adminUser: "admin"
+    adminPassword: ""  # If empty, authentication is disabled.
+
+  # Data directory for admin configuration and maintenance data
+  dataDir: ""  # If empty, configuration is kept in memory only
+
+  # Master servers to connect to
+  # If empty, uses global.masterServer or auto-discovers from master statefulset
+  masters: ""
+
+  # Custom command line arguments to add to the admin command
+  # Example: ["-customFlag", "value", "-anotherFlag"]
+  extraArgs: []
+
+  # Storage configuration
+  data:
+    type: "emptyDir"  # Options: "hostPath", "persistentVolumeClaim", "emptyDir", "existingClaim"
+    size: "10Gi"
+    storageClass: ""
+    hostPathPrefix: /storage
+    claimName: ""
+    annotations: {}
+
+  logs:
+    type: "emptyDir"  # Options: "hostPath", "persistentVolumeClaim", "emptyDir", "existingClaim"
+    size: "5Gi"
+    storageClass: ""
+    hostPathPrefix: /storage
+    claimName: ""
+    annotations: {}
+
+  # Additional resources
+  sidecars: []
+  initContainers: ""
+  extraVolumes: ""
+  extraVolumeMounts: ""
+  podLabels: {}
+  podAnnotations: {}
+  annotations: {}
+
+  ## Set podManagementPolicy
+  podManagementPolicy: Parallel
+
+  # Affinity Settings
+  # Commenting out or setting as empty the affinity variable, will allow
+  # deployment to single node services such as Minikube
+  affinity: |
+    podAntiAffinity:
+      requiredDuringSchedulingIgnoredDuringExecution:
+        - labelSelector:
+            matchLabels:
+              app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+              app.kubernetes.io/instance: {{ .Release.Name }}
+              app.kubernetes.io/component: admin
+          topologyKey: kubernetes.io/hostname
+
+  # Topology Spread Constraints Settings
+  # This should map directly to the value of the topologySpreadConstraints
+  # for a PodSpec. By Default no constraints are set.
+  topologySpreadConstraints: ""
+
+  resources: {}
+  tolerations: ""
+  nodeSelector: ""
+  priorityClassName: ""
+  serviceAccountName: ""
+  podSecurityContext: {}
+  containerSecurityContext: {}
+
+  extraEnvironmentVars: {}
+
+  # Health checks
+  livenessProbe:
+    enabled: true
+    httpGet:
+      path: /health
+      scheme: HTTP
+    initialDelaySeconds: 20
+    periodSeconds: 60
+    successThreshold: 1
+    failureThreshold: 5
+    timeoutSeconds: 10
+
+  readinessProbe:
+    enabled: true
+    httpGet:
+      path: /health
+      scheme: HTTP
+    initialDelaySeconds: 15
+    periodSeconds: 15
+    successThreshold: 1
+    failureThreshold: 3
+    timeoutSeconds: 10
+
+  ingress:
+    enabled: false
+    className: "nginx"
+    # host: false for "*" hostname
+    host: "admin.seaweedfs.local"
+    path: "/"
+    pathType: Prefix
+    annotations: {}
+    tls: []
+
+  service:
+    type: ClusterIP
+    annotations: {}
+
+  # ServiceMonitor annotations (separate from pod/deployment annotations)
+  serviceMonitor:
+    annotations: {}
+
+worker:
+  enabled: false
+  imageOverride: null
+  restartPolicy: null
+  replicas: 1
+  loggingOverrideLevel: null
+  metricsPort: 9327
+
+  # Admin server to connect to
+  # Format: "host:port" or auto-discover from admin service
+  adminServer: ""
+
+  # Worker capabilities - comma-separated list
+  # Available: vacuum, balance, erasure_coding
+  # Default: "vacuum,balance,erasure_coding" (all capabilities)
+  capabilities: "vacuum,balance,erasure_coding"
+
+  # Maximum number of concurrent tasks
+  maxConcurrent: 3
+
+  # Working directory for task execution
+  workingDir: "/tmp/seaweedfs-worker"
+
+  # Custom command line arguments to add to the worker command
+  # Example: ["-customFlag", "value", "-anotherFlag"]
+  extraArgs: []
+
+  # Storage configuration for working directory
+  # Note: Workers use Deployment, so use "emptyDir", "hostPath", or "existingClaim"
+  # Do NOT use "persistentVolumeClaim" - use "existingClaim" with pre-provisioned PVC instead
+  data:
+    type: "emptyDir"  # Options: "hostPath", "emptyDir", "existingClaim"
+    hostPathPrefix: /storage
+    claimName: ""  # For existingClaim type
+
+  logs:
+    type: "emptyDir"  # Options: "hostPath", "emptyDir", "existingClaim"
+    hostPathPrefix: /storage
+    claimName: ""  # For existingClaim type
+
+  # Additional resources
+  sidecars: []
+  initContainers: ""
+  extraVolumes: ""
+  extraVolumeMounts: ""
+  podLabels: {}
+  podAnnotations: {}
+  annotations: {}
+
+  # Affinity Settings
+  # Commenting out or setting as empty the affinity variable, will allow
+  # deployment to single node services such as Minikube
+  affinity: |
+    podAntiAffinity:
+      requiredDuringSchedulingIgnoredDuringExecution:
+        - labelSelector:
+            matchLabels:
+              app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+              app.kubernetes.io/instance: {{ .Release.Name }}
+              app.kubernetes.io/component: worker
+          topologyKey: kubernetes.io/hostname
+
+  # Topology Spread Constraints Settings
+  # This should map directly to the value of the topologySpreadConstraints
+  # for a PodSpec. By Default no constraints are set.
+  topologySpreadConstraints: ""
+
+  resources:
+    requests:
+      cpu: "500m"
+      memory: "512Mi"
+    limits:
+      cpu: "2"
+      memory: "2Gi"
+  tolerations: ""
+  nodeSelector: ""
+  priorityClassName: ""
+  serviceAccountName: ""
+  podSecurityContext: {}
+  containerSecurityContext: {}
+
+  extraEnvironmentVars: {}
+
+  # Health checks for worker pods
+  # Since workers do not have an HTTP endpoint, a tcpSocket probe on the metrics port is recommended.
+  livenessProbe:
+    enabled: true
+    tcpSocket:
+      port: metrics
+    initialDelaySeconds: 30
+    periodSeconds: 60
+    successThreshold: 1
+    failureThreshold: 5
+    timeoutSeconds: 10
+
+  readinessProbe:
+    enabled: true
+    tcpSocket:
+      port: metrics
+    initialDelaySeconds: 20
+    periodSeconds: 15
+    successThreshold: 1
+    failureThreshold: 3
+    timeoutSeconds: 10
+
+  # ServiceMonitor annotations (separate from pod/deployment annotations)
+  serviceMonitor:
+    annotations: {}
+
 # All-in-one deployment configuration
 allInOne:
  enabled: false
@@ -0,0 +1,27 @@
+.PHONY: test test-short build-weed clean
+
+# Build weed binary first if needed
+build-weed:
+	cd ../../weed && go build -o weed .
+
+# Run all integration tests
+test: build-weed
+	go test -v -timeout 5m .
+
+# Run tests in short mode (skip integration tests)
+test-short:
+	go test -v -short .
+
+# Run specific test
+test-basic: build-weed
+	go test -v -timeout 3m -run TestMetadataSubscribeBasic .
+
+test-stall: build-weed
+	go test -v -timeout 5m -run TestMetadataSubscribeSingleFilerNoStall .
+
+test-resume: build-weed
+	go test -v -timeout 3m -run TestMetadataSubscribeResumeFromDisk .
+
+clean:
+	rm -f ../../weed/weed
+
@@ -0,0 +1,49 @@
+# Metadata Subscribe Integration Tests
+
+This directory contains integration tests for the SeaweedFS metadata subscription functionality.
+
+## Tests
+
+### TestMetadataSubscribeBasic
+Tests basic metadata subscription functionality:
+- Start a SeaweedFS cluster (master, volume, filer)
+- Subscribe to metadata changes
+- Upload files and verify events are received
+
+### TestMetadataSubscribeSingleFilerNoStall
+Regression test for [issue #4977](https://github.com/seaweedfs/seaweedfs/issues/4977):
+- Tests that metadata subscription doesn't stall in single-filer setups
+- Simulates high-load file uploads while a subscriber tries to keep up
+- Verifies that events are received without significant stalling
+
+The bug was that in single-filer setups, `SubscribeMetadata` would block indefinitely
+on `MetaAggregator.MetaLogBuffer` which remains empty (no peers to aggregate from).
+The fix ensures that when the buffer is empty, the subscription returns to read from
+persisted logs on disk.
+
+### TestMetadataSubscribeResumeFromDisk
+Tests that subscription can resume from disk:
+- Upload files before starting subscription
+- Wait for logs to be flushed to disk
+- Start subscription from the beginning
+- Verify pre-uploaded files are received from disk
+
+## Running Tests
+
+```bash
+# Run all tests (requires weed binary in PATH or built)
+go test -v ./test/metadata_subscribe/...
+
+# Skip integration tests
+go test -short ./test/metadata_subscribe/...
+
+# Run with increased timeout for slow systems
+go test -v -timeout 5m ./test/metadata_subscribe/...
+```
+
+## Requirements
+
+- `weed` binary must be available in PATH or in the parent directories
+- Tests create temporary directories that are cleaned up after completion
+- Tests use ports 9333 (master), 8080 (volume), 8888 (filer)
+
@@ -0,0 +1,917 @@
+package metadata_subscribe
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// TestMetadataSubscribeBasic tests basic metadata subscription functionality
+func TestMetadataSubscribeBasic(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	testDir, err := os.MkdirTemp("", "seaweedfs_metadata_subscribe_test_")
+	require.NoError(t, err)
+	defer os.RemoveAll(testDir)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	cluster, err := startSeaweedFSCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	// Wait for servers to be ready
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:9333", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8080", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8888", 30*time.Second))
+
+	t.Logf("SeaweedFS cluster started successfully")
+
+	t.Run("subscribe_and_receive_events", func(t *testing.T) {
+		// Create a channel to receive events
+		eventsChan := make(chan *filer_pb.SubscribeMetadataResponse, 100)
+		errChan := make(chan error, 1)
+
+		// Start subscribing in a goroutine
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		go func() {
+			err := subscribeToMetadata(subCtx, "127.0.0.1:8888", "/", eventsChan)
+			if err != nil && !strings.Contains(err.Error(), "context canceled") {
+				errChan <- err
+			}
+		}()
+
+		// Wait for subscription to be established
+		time.Sleep(2 * time.Second)
+
+		// Create test files via HTTP
+		testFiles := []string{
+			"/test/file1.txt",
+			"/test/file2.txt",
+			"/test/subdir/file3.txt",
+		}
+
+		for _, path := range testFiles {
+			err := uploadFile("http://127.0.0.1:8888"+path, []byte("test content for "+path))
+			require.NoError(t, err, "Failed to upload %s", path)
+			t.Logf("Uploaded %s", path)
+		}
+
+		// Collect events with timeout
+		receivedPaths := make(map[string]bool)
+		timeout := time.After(30 * time.Second)
+
+	eventLoop:
+		for {
+			select {
+			case event := <-eventsChan:
+				if event.EventNotification != nil && event.EventNotification.NewEntry != nil {
+					path := filepath.Join(event.Directory, event.EventNotification.NewEntry.Name)
+					t.Logf("Received event for: %s", path)
+					receivedPaths[path] = true
+				}
+				// Check if we received all expected events
+				allReceived := true
+				for _, p := range testFiles {
+					if !receivedPaths[p] {
+						allReceived = false
+						break
+					}
+				}
+				if allReceived {
+					break eventLoop
+				}
+			case err := <-errChan:
+				t.Fatalf("Subscription error: %v", err)
+			case <-timeout:
+				t.Logf("Timeout waiting for events. Received %d/%d events", len(receivedPaths), len(testFiles))
+				break eventLoop
+			}
+		}
+
+		// Verify we received events for all test files
+		for _, path := range testFiles {
+			assert.True(t, receivedPaths[path], "Should have received event for %s", path)
+		}
+	})
+}
+
+// TestMetadataSubscribeSingleFilerNoStall tests that subscription doesn't stall
+// in single-filer setups (regression test for issue #4977)
+func TestMetadataSubscribeSingleFilerNoStall(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	testDir, err := os.MkdirTemp("", "seaweedfs_single_filer_test_")
+	require.NoError(t, err)
+	defer os.RemoveAll(testDir)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second)
+	defer cancel()
+
+	cluster, err := startSeaweedFSCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	// Wait for servers to be ready
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:9333", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8080", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8888", 30*time.Second))
+
+	t.Logf("Single-filer cluster started")
+
+	t.Run("high_load_subscription_no_stall", func(t *testing.T) {
+		// This test simulates the scenario from issue #4977:
+		// High-load writes while a subscriber tries to keep up
+
+		var receivedCount int64
+		var uploadedCount int64
+
+		eventsChan := make(chan *filer_pb.SubscribeMetadataResponse, 1000)
+		errChan := make(chan error, 1)
+
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		// Start subscriber
+		go func() {
+			err := subscribeToMetadata(subCtx, "127.0.0.1:8888", "/", eventsChan)
+			if err != nil && !strings.Contains(err.Error(), "context canceled") {
+				errChan <- err
+			}
+		}()
+
+		// Wait for subscription to be established
+		time.Sleep(2 * time.Second)
+
+		// Start counting received events
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case event := <-eventsChan:
+					if event.EventNotification != nil && event.EventNotification.NewEntry != nil {
+						if !event.EventNotification.NewEntry.IsDirectory {
+							atomic.AddInt64(&receivedCount, 1)
+						}
+					}
+				case <-subCtx.Done():
+					return
+				}
+			}
+		}()
+
+		// Upload files concurrently (simulate high load)
+		numFiles := 100
+		numWorkers := 10
+
+		uploadWg := sync.WaitGroup{}
+		for w := 0; w < numWorkers; w++ {
+			uploadWg.Add(1)
+			go func(workerId int) {
+				defer uploadWg.Done()
+				for i := 0; i < numFiles/numWorkers; i++ {
+					path := fmt.Sprintf("/load_test/worker%d/file%d.txt", workerId, i)
+					err := uploadFile("http://127.0.0.1:8888"+path, []byte(fmt.Sprintf("content %d-%d", workerId, i)))
+					if err == nil {
+						atomic.AddInt64(&uploadedCount, 1)
+					}
+				}
+			}(w)
+		}
+
+		uploadWg.Wait()
+		uploaded := atomic.LoadInt64(&uploadedCount)
+		t.Logf("Uploaded %d files", uploaded)
+
+		// Wait for events to be received (with timeout to detect stall)
+		stallTimeout := time.After(60 * time.Second)
+		checkInterval := time.NewTicker(2 * time.Second)
+		defer checkInterval.Stop()
+
+		lastReceived := atomic.LoadInt64(&receivedCount)
+		staleCount := 0
+
+	waitLoop:
+		for {
+			select {
+			case <-stallTimeout:
+				received := atomic.LoadInt64(&receivedCount)
+				t.Logf("Timeout: received %d/%d events (%.1f%%)",
+					received, uploaded, float64(received)/float64(uploaded)*100)
+				break waitLoop
+			case <-checkInterval.C:
+				received := atomic.LoadInt64(&receivedCount)
+				if received >= uploaded {
+					t.Logf("All %d events received", received)
+					break waitLoop
+				}
+				if received == lastReceived {
+					staleCount++
+					if staleCount >= 5 {
+						// If no progress for 10 seconds, subscription may be stalled
+						t.Logf("WARNING: No progress for %d checks. Received %d/%d (%.1f%%)",
+							staleCount, received, uploaded, float64(received)/float64(uploaded)*100)
+					}
+				} else {
+					staleCount = 0
+					t.Logf("Progress: received %d/%d events (%.1f%%)",
+						received, uploaded, float64(received)/float64(uploaded)*100)
+				}
+				lastReceived = received
+			case err := <-errChan:
+				t.Fatalf("Subscription error: %v", err)
+			}
+		}
+
+		subCancel()
+		wg.Wait()
+
+		received := atomic.LoadInt64(&receivedCount)
+
+		// With the fix for #4977, we should receive a high percentage of events
+		// Before the fix, this would stall at ~20-40%
+		percentage := float64(received) / float64(uploaded) * 100
+		t.Logf("Final: received %d/%d events (%.1f%%)", received, uploaded, percentage)
+
+		// We should receive at least 80% of events (allowing for some timing issues)
+		assert.GreaterOrEqual(t, percentage, 80.0,
+			"Should receive at least 80%% of events (received %.1f%%)", percentage)
+	})
+}
+
+// TestMetadataSubscribeResumeFromDisk tests that subscription can resume from disk
+func TestMetadataSubscribeResumeFromDisk(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	testDir, err := os.MkdirTemp("", "seaweedfs_resume_test_")
+	require.NoError(t, err)
+	defer os.RemoveAll(testDir)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	cluster, err := startSeaweedFSCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:9333", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8080", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8888", 30*time.Second))
+
+	t.Run("upload_before_subscribe", func(t *testing.T) {
+		// Upload files BEFORE starting subscription
+		numFiles := 20
+		for i := 0; i < numFiles; i++ {
+			path := fmt.Sprintf("/pre_subscribe/file%d.txt", i)
+			err := uploadFile("http://127.0.0.1:8888"+path, []byte(fmt.Sprintf("content %d", i)))
+			require.NoError(t, err)
+		}
+		t.Logf("Uploaded %d files before subscription", numFiles)
+
+		// Wait for logs to be flushed to disk
+		time.Sleep(15 * time.Second)
+
+		// Now start subscription from the beginning
+		eventsChan := make(chan *filer_pb.SubscribeMetadataResponse, 100)
+		errChan := make(chan error, 1)
+
+		subCtx, subCancel := context.WithTimeout(ctx, 30*time.Second)
+		defer subCancel()
+
+		go func() {
+			err := subscribeToMetadataFromBeginning(subCtx, "127.0.0.1:8888", "/pre_subscribe/", eventsChan)
+			if err != nil && !strings.Contains(err.Error(), "context") {
+				errChan <- err
+			}
+		}()
+
+		// Count received events
+		receivedCount := 0
+		timeout := time.After(30 * time.Second)
+
+	countLoop:
+		for {
+			select {
+			case event := <-eventsChan:
+				if event.EventNotification != nil && event.EventNotification.NewEntry != nil {
+					if !event.EventNotification.NewEntry.IsDirectory {
+						receivedCount++
+						t.Logf("Received event %d: %s/%s", receivedCount,
+							event.Directory, event.EventNotification.NewEntry.Name)
+					}
+				}
+				if receivedCount >= numFiles {
+					break countLoop
+				}
+			case err := <-errChan:
+				t.Fatalf("Subscription error: %v", err)
+			case <-timeout:
+				t.Logf("Timeout: received %d/%d events", receivedCount, numFiles)
+				break countLoop
+			}
+		}
+
+		// Should receive all pre-uploaded files from disk
+		assert.GreaterOrEqual(t, receivedCount, numFiles-2, // Allow small margin
+			"Should receive most pre-uploaded files from disk (received %d/%d)", receivedCount, numFiles)
+	})
+}
+
+// TestMetadataSubscribeConcurrentWrites tests subscription with concurrent writes
+func TestMetadataSubscribeConcurrentWrites(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	testDir, err := os.MkdirTemp("", "seaweedfs_concurrent_writes_test_")
+	require.NoError(t, err)
+	defer os.RemoveAll(testDir)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second)
+	defer cancel()
+
+	cluster, err := startSeaweedFSCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:9333", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8080", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8888", 30*time.Second))
+
+	t.Logf("Cluster started for concurrent writes test")
+
+	t.Run("concurrent_goroutine_writes", func(t *testing.T) {
+		var receivedCount int64
+		var uploadedCount int64
+
+		eventsChan := make(chan *filer_pb.SubscribeMetadataResponse, 10000)
+		errChan := make(chan error, 1)
+
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		// Start subscriber
+		go func() {
+			err := subscribeToMetadata(subCtx, "127.0.0.1:8888", "/concurrent/", eventsChan)
+			if err != nil && !strings.Contains(err.Error(), "context") {
+				errChan <- err
+			}
+		}()
+
+		time.Sleep(2 * time.Second)
+
+		// Start counting received events
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case event := <-eventsChan:
+					if event.EventNotification != nil && event.EventNotification.NewEntry != nil {
+						if !event.EventNotification.NewEntry.IsDirectory {
+							atomic.AddInt64(&receivedCount, 1)
+						}
+					}
+				case <-subCtx.Done():
+					return
+				}
+			}
+		}()
+
+		// Launch many concurrent writers
+		numWorkers := 50
+		filesPerWorker := 20
+		totalExpected := int64(numWorkers * filesPerWorker)
+
+		uploadWg := sync.WaitGroup{}
+		for w := 0; w < numWorkers; w++ {
+			uploadWg.Add(1)
+			go func(workerId int) {
+				defer uploadWg.Done()
+				for i := 0; i < filesPerWorker; i++ {
+					path := fmt.Sprintf("/concurrent/w%d/f%d.txt", workerId, i)
+					content := []byte(fmt.Sprintf("worker%d-file%d", workerId, i))
+					if err := uploadFile("http://127.0.0.1:8888"+path, content); err == nil {
+						atomic.AddInt64(&uploadedCount, 1)
+					}
+				}
+			}(w)
+		}
+
+		uploadWg.Wait()
+		uploaded := atomic.LoadInt64(&uploadedCount)
+		t.Logf("Uploaded %d/%d files from %d concurrent workers", uploaded, totalExpected, numWorkers)
+
+		// Wait for events with progress tracking
+		stallTimeout := time.After(90 * time.Second)
+		checkInterval := time.NewTicker(3 * time.Second)
+		defer checkInterval.Stop()
+
+		lastReceived := int64(0)
+		stableCount := 0
+
+	waitLoop:
+		for {
+			select {
+			case <-stallTimeout:
+				break waitLoop
+			case <-checkInterval.C:
+				received := atomic.LoadInt64(&receivedCount)
+				if received >= uploaded {
+					t.Logf("All %d events received", received)
+					break waitLoop
+				}
+				if received == lastReceived {
+					stableCount++
+					if stableCount >= 5 {
+						t.Logf("No progress for %d checks, received %d/%d", stableCount, received, uploaded)
+						break waitLoop
+					}
+				} else {
+					stableCount = 0
+					t.Logf("Progress: %d/%d (%.1f%%)", received, uploaded, float64(received)/float64(uploaded)*100)
+				}
+				lastReceived = received
+			case err := <-errChan:
+				t.Fatalf("Subscription error: %v", err)
+			}
+		}
+
+		subCancel()
+		wg.Wait()
+
+		received := atomic.LoadInt64(&receivedCount)
+		percentage := float64(received) / float64(uploaded) * 100
+		t.Logf("Final: received %d/%d events (%.1f%%)", received, uploaded, percentage)
+
+		// Should receive at least 80% of events
+		assert.GreaterOrEqual(t, percentage, 80.0,
+			"Should receive at least 80%% of concurrent write events")
+	})
+}
+
+// TestMetadataSubscribeMillionUpdates tests subscription with 1 million metadata updates
+// This test creates metadata entries directly via gRPC without actual file content
+func TestMetadataSubscribeMillionUpdates(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	testDir, err := os.MkdirTemp("", "seaweedfs_million_updates_test_")
+	require.NoError(t, err)
+	defer os.RemoveAll(testDir)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+
+	cluster, err := startSeaweedFSCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:9333", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8080", 30*time.Second))
+	require.NoError(t, waitForHTTPServer("http://127.0.0.1:8888", 30*time.Second))
+
+	t.Logf("Cluster started for million updates test")
+
+	t.Run("million_metadata_updates", func(t *testing.T) {
+		var receivedCount int64
+		var createdCount int64
+		totalEntries := int64(1_000_000)
+
+		eventsChan := make(chan *filer_pb.SubscribeMetadataResponse, 100000)
+		errChan := make(chan error, 1)
+
+		subCtx, subCancel := context.WithCancel(ctx)
+		defer subCancel()
+
+		// Start subscriber
+		go func() {
+			err := subscribeToMetadata(subCtx, "127.0.0.1:8888", "/million/", eventsChan)
+			if err != nil && !strings.Contains(err.Error(), "context") {
+				errChan <- err
+			}
+		}()
+
+		time.Sleep(2 * time.Second)
+
+		// Start counting received events
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case event := <-eventsChan:
+					if event.EventNotification != nil && event.EventNotification.NewEntry != nil {
+						if !event.EventNotification.NewEntry.IsDirectory {
+							atomic.AddInt64(&receivedCount, 1)
+						}
+					}
+				case <-subCtx.Done():
+					return
+				}
+			}
+		}()
+
+		// Create metadata entries directly via gRPC (no actual file content)
+		numWorkers := 100
+		entriesPerWorker := int(totalEntries) / numWorkers
+
+		startTime := time.Now()
+		createWg := sync.WaitGroup{}
+
+		for w := 0; w < numWorkers; w++ {
+			createWg.Add(1)
+			go func(workerId int) {
+				defer createWg.Done()
+				grpcDialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+				err := pb.WithFilerClient(false, 0, pb.ServerAddress("127.0.0.1:8888"), grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+					for i := 0; i < entriesPerWorker; i++ {
+						dir := fmt.Sprintf("/million/bucket%d", workerId%100)
+						name := fmt.Sprintf("entry_%d_%d", workerId, i)
+
+						_, err := client.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
+							Directory: dir,
+							Entry: &filer_pb.Entry{
+								Name:        name,
+								IsDirectory: false,
+								Attributes: &filer_pb.FuseAttributes{
+									FileSize: 100,
+									Mtime:    time.Now().Unix(),
+									FileMode: 0644,
+									Uid:      1000,
+									Gid:      1000,
+								},
+							},
+						})
+						if err == nil {
+							atomic.AddInt64(&createdCount, 1)
+						}
+
+						// Log progress every 10000 entries per worker
+						if i > 0 && i%10000 == 0 {
+							created := atomic.LoadInt64(&createdCount)
+							elapsed := time.Since(startTime)
+							rate := float64(created) / elapsed.Seconds()
+							t.Logf("Worker %d: created %d entries, total %d (%.0f/sec)",
+								workerId, i, created, rate)
+						}
+					}
+					return nil
+				})
+				if err != nil {
+					t.Logf("Worker %d error: %v", workerId, err)
+				}
+			}(w)
+		}
+
+		// Progress reporter
+		progressDone := make(chan struct{})
+		go func() {
+			ticker := time.NewTicker(10 * time.Second)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ticker.C:
+					created := atomic.LoadInt64(&createdCount)
+					received := atomic.LoadInt64(&receivedCount)
+					elapsed := time.Since(startTime)
+					createRate := float64(created) / elapsed.Seconds()
+					receiveRate := float64(received) / elapsed.Seconds()
+					t.Logf("Progress: created %d (%.0f/sec), received %d (%.0f/sec), lag %d",
+						created, createRate, received, receiveRate, created-received)
+				case <-progressDone:
+					return
+				}
+			}
+		}()
+
+		createWg.Wait()
+		close(progressDone)
+
+		created := atomic.LoadInt64(&createdCount)
+		elapsed := time.Since(startTime)
+		t.Logf("Created %d entries in %v (%.0f/sec)", created, elapsed, float64(created)/elapsed.Seconds())
+
+		// Wait for subscription to catch up
+		catchupTimeout := time.After(5 * time.Minute)
+		checkInterval := time.NewTicker(5 * time.Second)
+		defer checkInterval.Stop()
+
+		lastReceived := int64(0)
+		stableCount := 0
+
+	waitLoop:
+		for {
+			select {
+			case <-catchupTimeout:
+				t.Logf("Catchup timeout reached")
+				break waitLoop
+			case <-checkInterval.C:
+				received := atomic.LoadInt64(&receivedCount)
+				if received >= created {
+					t.Logf("All %d events received", received)
+					break waitLoop
+				}
+				if received == lastReceived {
+					stableCount++
+					if stableCount >= 10 {
+						t.Logf("No progress for %d checks", stableCount)
+						break waitLoop
+					}
+				} else {
+					stableCount = 0
+					rate := float64(received-lastReceived) / 5.0
+					t.Logf("Catching up: %d/%d (%.1f%%) at %.0f/sec",
+						received, created, float64(received)/float64(created)*100, rate)
+				}
+				lastReceived = received
+			case err := <-errChan:
+				t.Fatalf("Subscription error: %v", err)
+			}
+		}
+
+		subCancel()
+		wg.Wait()
+
+		received := atomic.LoadInt64(&receivedCount)
+		percentage := float64(received) / float64(created) * 100
+		totalTime := time.Since(startTime)
+		t.Logf("Final: created %d, received %d (%.1f%%) in %v", created, received, percentage, totalTime)
+
+		// For million entries, we expect at least 90% to be received
+		assert.GreaterOrEqual(t, percentage, 90.0,
+			"Should receive at least 90%% of million metadata events (received %.1f%%)", percentage)
+	})
+}
+
+// Helper types and functions
+
+type TestCluster struct {
+	masterCmd *exec.Cmd
+	volumeCmd *exec.Cmd
+	filerCmd  *exec.Cmd
+	testDir   string
+}
+
+func (c *TestCluster) Stop() {
+	if c.filerCmd != nil && c.filerCmd.Process != nil {
+		c.filerCmd.Process.Kill()
+		c.filerCmd.Wait()
+	}
+	if c.volumeCmd != nil && c.volumeCmd.Process != nil {
+		c.volumeCmd.Process.Kill()
+		c.volumeCmd.Wait()
+	}
+	if c.masterCmd != nil && c.masterCmd.Process != nil {
+		c.masterCmd.Process.Kill()
+		c.masterCmd.Wait()
+	}
+}
+
+func startSeaweedFSCluster(ctx context.Context, dataDir string) (*TestCluster, error) {
+	weedBinary := findWeedBinary()
+	if weedBinary == "" {
+		return nil, fmt.Errorf("weed binary not found")
+	}
+
+	cluster := &TestCluster{testDir: dataDir}
+
+	// Create directories
+	masterDir := filepath.Join(dataDir, "master")
+	volumeDir := filepath.Join(dataDir, "volume")
+	filerDir := filepath.Join(dataDir, "filer")
+	if err := os.MkdirAll(masterDir, 0755); err != nil {
+		return nil, fmt.Errorf("failed to create master dir: %v", err)
+	}
+	if err := os.MkdirAll(volumeDir, 0755); err != nil {
+		return nil, fmt.Errorf("failed to create volume dir: %v", err)
+	}
+	if err := os.MkdirAll(filerDir, 0755); err != nil {
+		return nil, fmt.Errorf("failed to create filer dir: %v", err)
+	}
+
+	// Start master server
+	masterCmd := exec.CommandContext(ctx, weedBinary, "master",
+		"-port", "9333",
+		"-mdir", masterDir,
+		"-volumeSizeLimitMB", "10",
+		"-ip", "127.0.0.1",
+		"-peers", "none",
+	)
+	masterLogFile, err := os.Create(filepath.Join(masterDir, "master.log"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create master log file: %v", err)
+	}
+	masterCmd.Stdout = masterLogFile
+	masterCmd.Stderr = masterLogFile
+	if err := masterCmd.Start(); err != nil {
+		return nil, fmt.Errorf("failed to start master: %v", err)
+	}
+	cluster.masterCmd = masterCmd
+
+	time.Sleep(2 * time.Second)
+
+	// Start volume server
+	volumeCmd := exec.CommandContext(ctx, weedBinary, "volume",
+		"-port", "8080",
+		"-dir", volumeDir,
+		"-max", "10",
+		"-master", "127.0.0.1:9333",
+		"-ip", "127.0.0.1",
+	)
+	volumeLogFile, err := os.Create(filepath.Join(volumeDir, "volume.log"))
+	if err != nil {
+		cluster.Stop()
+		return nil, fmt.Errorf("failed to create volume log file: %v", err)
+	}
+	volumeCmd.Stdout = volumeLogFile
+	volumeCmd.Stderr = volumeLogFile
+	if err := volumeCmd.Start(); err != nil {
+		cluster.Stop()
+		return nil, fmt.Errorf("failed to start volume: %v", err)
+	}
+	cluster.volumeCmd = volumeCmd
+
+	time.Sleep(2 * time.Second)
+
+	// Start filer server
+	filerCmd := exec.CommandContext(ctx, weedBinary, "filer",
+		"-port", "8888",
+		"-master", "127.0.0.1:9333",
+		"-ip", "127.0.0.1",
+	)
+	filerLogFile, err := os.Create(filepath.Join(filerDir, "filer.log"))
+	if err != nil {
+		cluster.Stop()
+		return nil, fmt.Errorf("failed to create filer log file: %v", err)
+	}
+	filerCmd.Stdout = filerLogFile
+	filerCmd.Stderr = filerLogFile
+	if err := filerCmd.Start(); err != nil {
+		cluster.Stop()
+		return nil, fmt.Errorf("failed to start filer: %v", err)
+	}
+	cluster.filerCmd = filerCmd
+
+	time.Sleep(3 * time.Second)
+
+	return cluster, nil
+}
+
+func findWeedBinary() string {
+	candidates := []string{
+		"../../../weed/weed",
+		"../../weed/weed",
+		"./weed",
+		"weed",
+	}
+	for _, candidate := range candidates {
+		if _, err := os.Stat(candidate); err == nil {
+			return candidate
+		}
+	}
+	if path, err := exec.LookPath("weed"); err == nil {
+		return path
+	}
+	return ""
+}
+
+func waitForHTTPServer(url string, timeout time.Duration) error {
+	start := time.Now()
+	for time.Since(start) < timeout {
+		resp, err := http.Get(url)
+		if err == nil {
+			resp.Body.Close()
+			return nil
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	return fmt.Errorf("timeout waiting for server %s", url)
+}
+
+func uploadFile(url string, content []byte) error {
+	// Create multipart form data
+	var buf bytes.Buffer
+	writer := multipart.NewWriter(&buf)
+
+	// Extract filename from URL path
+	parts := strings.Split(url, "/")
+	filename := parts[len(parts)-1]
+	if filename == "" {
+		filename = "file.txt"
+	}
+
+	// Create form file field
+	part, err := writer.CreateFormFile("file", filename)
+	if err != nil {
+		return fmt.Errorf("create form file: %v", err)
+	}
+	if _, err := part.Write(content); err != nil {
+		return fmt.Errorf("write content: %v", err)
+	}
+	if err := writer.Close(); err != nil {
+		return fmt.Errorf("close writer: %v", err)
+	}
+
+	req, err := http.NewRequest("POST", url, &buf)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", writer.FormDataContentType())
+
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 400 {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("upload failed with status %d: %s", resp.StatusCode, string(body))
+	}
+	return nil
+}
+
+func subscribeToMetadata(ctx context.Context, filerGrpcAddress, pathPrefix string, eventsChan chan<- *filer_pb.SubscribeMetadataResponse) error {
+	return subscribeToMetadataWithOptions(ctx, filerGrpcAddress, pathPrefix, time.Now().UnixNano(), eventsChan)
+}
+
+func subscribeToMetadataFromBeginning(ctx context.Context, filerGrpcAddress, pathPrefix string, eventsChan chan<- *filer_pb.SubscribeMetadataResponse) error {
+	// Start from Unix epoch to get all events
+	return subscribeToMetadataWithOptions(ctx, filerGrpcAddress, pathPrefix, 0, eventsChan)
+}
+
+func subscribeToMetadataWithOptions(ctx context.Context, filerGrpcAddress, pathPrefix string, sinceNs int64, eventsChan chan<- *filer_pb.SubscribeMetadataResponse) error {
+	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
+	if grpcDialOption == nil {
+		grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
+	}
+
+	return pb.WithFilerClient(false, 0, pb.ServerAddress(filerGrpcAddress), grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
+			ClientName: "integration_test",
+			PathPrefix: pathPrefix,
+			SinceNs:    sinceNs,
+			ClientId:   util.RandomInt32(),
+		})
+		if err != nil {
+			return fmt.Errorf("subscribe: %v", err)
+		}
+
+		for {
+			resp, err := stream.Recv()
+			if err != nil {
+				if err == io.EOF || ctx.Err() != nil {
+					return nil
+				}
+				return err
+			}
+
+			select {
+			case eventsChan <- resp:
+			case <-ctx.Done():
+				return nil
+			case <-time.After(100 * time.Millisecond):
+				// Channel full after brief wait, log warning
+				glog.Warningf("Event channel full, skipping event for %s", resp.Directory)
+			}
+		}
+	})
+}
@@ -99,7 +99,7 @@ func init() {
 	f.metricsHttpIp = cmdFiler.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
 	f.saveToFilerLimit = cmdFiler.Flag.Int("saveToFilerLimit", 0, "files smaller than this limit will be saved in filer store")
 	f.defaultLevelDbDirectory = cmdFiler.Flag.String("defaultStoreDir", ".", "if filer.toml is empty, use an embedded filer store in the directory")
-	f.concurrentUploadLimitMB = cmdFiler.Flag.Int("concurrentUploadLimitMB", 128, "limit total concurrent upload size")
+	f.concurrentUploadLimitMB = cmdFiler.Flag.Int("concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited")
 	f.concurrentFileUploadLimit = cmdFiler.Flag.Int("concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited")
 	f.debug = cmdFiler.Flag.Bool("debug", false, "serves runtime profiling data, e.g., http://localhost:<debug.port>/debug/pprof/goroutine?debug=2")
 	f.debugPort = cmdFiler.Flag.Int("debug.port", 6060, "http port for debugging")
@@ -132,7 +132,7 @@ func init() {
 	filerS3Options.tlsVerifyClientCert = cmdFiler.Flag.Bool("s3.tlsVerifyClientCert", false, "whether to verify the client's certificate")
 	filerS3Options.bindIp = cmdFiler.Flag.String("s3.ip.bind", "", "ip address to bind to. If empty, default to same as -ip.bind option.")
 	filerS3Options.idleTimeout = cmdFiler.Flag.Int("s3.idleTimeout", 120, "connection idle seconds")
-	filerS3Options.concurrentUploadLimitMB = cmdFiler.Flag.Int("s3.concurrentUploadLimitMB", 128, "limit total concurrent upload size for S3")
+	filerS3Options.concurrentUploadLimitMB = cmdFiler.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size for S3, 0 means unlimited")
 	filerS3Options.concurrentFileUploadLimit = cmdFiler.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited")

 	// start webdav on filer
@@ -37,6 +37,9 @@ type MountOptions struct {
 	extraOptions       []string
 	fuseCommandPid     int

+	// Periodic metadata flush to protect against orphan chunk cleanup
+	metadataFlushSeconds *int
+
 	// RDMA acceleration options
 	rdmaEnabled       *bool
 	rdmaSidecarAddr   *string
@@ -85,6 +88,9 @@ func init() {
 	mountOptions.disableXAttr = cmdMount.Flag.Bool("disableXAttr", false, "disable xattr")
 	mountOptions.fuseCommandPid = 0

+	// Periodic metadata flush to protect against orphan chunk cleanup
+	mountOptions.metadataFlushSeconds = cmdMount.Flag.Int("metadataFlushSeconds", 120, "periodically flush file metadata to filer in seconds (0 to disable). This protects chunks from being purged by volume.fsck for long-running writes")
+
 	// RDMA acceleration flags
 	mountOptions.rdmaEnabled = cmdMount.Flag.Bool("rdma.enabled", false, "enable RDMA acceleration for reads")
 	mountOptions.rdmaSidecarAddr = cmdMount.Flag.String("rdma.sidecar", "", "RDMA sidecar address (e.g., localhost:8081)")
@@ -254,6 +254,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		UidGidMapper:       uidGidMapper,
 		DisableXAttr:       *option.disableXAttr,
 		IsMacOs:            runtime.GOOS == "darwin",
+		MetadataFlushSeconds: *option.metadataFlushSeconds,
 		// RDMA acceleration options
 		RdmaEnabled:       *option.rdmaEnabled,
 		RdmaSidecarAddr:   *option.rdmaSidecarAddr,
@@ -84,7 +84,7 @@ func init() {
 	s3StandaloneOptions.localFilerSocket = cmdS3.Flag.String("localFilerSocket", "", "local filer socket path")
 	s3StandaloneOptions.localSocket = cmdS3.Flag.String("localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
 	s3StandaloneOptions.idleTimeout = cmdS3.Flag.Int("idleTimeout", 120, "connection idle seconds")
-	s3StandaloneOptions.concurrentUploadLimitMB = cmdS3.Flag.Int("concurrentUploadLimitMB", 128, "limit total concurrent upload size")
+	s3StandaloneOptions.concurrentUploadLimitMB = cmdS3.Flag.Int("concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited")
 	s3StandaloneOptions.concurrentFileUploadLimit = cmdS3.Flag.Int("concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited")
 }

@@ -122,7 +122,7 @@ func init() {
 	filerOptions.dirListingLimit = cmdServer.Flag.Int("filer.dirListLimit", 1000, "limit sub dir listing size")
 	filerOptions.cipher = cmdServer.Flag.Bool("filer.encryptVolumeData", false, "encrypt data on volume servers")
 	filerOptions.saveToFilerLimit = cmdServer.Flag.Int("filer.saveToFilerLimit", 0, "Small files smaller than this limit can be cached in filer store.")
-	filerOptions.concurrentUploadLimitMB = cmdServer.Flag.Int("filer.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
+	filerOptions.concurrentUploadLimitMB = cmdServer.Flag.Int("filer.concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited")
 	filerOptions.concurrentFileUploadLimit = cmdServer.Flag.Int("filer.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited")
 	filerOptions.localSocket = cmdServer.Flag.String("filer.localSocket", "", "default to /tmp/seaweedfs-filer-<port>.sock")
 	filerOptions.showUIDirectoryDelete = cmdServer.Flag.Bool("filer.ui.deleteDir", true, "enable filer UI show delete directory button")
@@ -142,8 +142,8 @@ func init() {
 	serverOptions.v.maintenanceMBPerSecond = cmdServer.Flag.Int("volume.maintenanceMBps", 0, "limit maintenance (replication / balance) IO rate in MB/s. Unset is 0, no limitation.")
 	serverOptions.v.fileSizeLimitMB = cmdServer.Flag.Int("volume.fileSizeLimitMB", 256, "limit file size to avoid out of memory")
 	serverOptions.v.ldbTimeout = cmdServer.Flag.Int64("volume.index.leveldbTimeout", 0, "alive time for leveldb (default to 0). If leveldb of volume is not accessed in ldbTimeout hours, it will be off loaded to reduce opened files and memory consumption.")
-	serverOptions.v.concurrentUploadLimitMB = cmdServer.Flag.Int("volume.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
-	serverOptions.v.concurrentDownloadLimitMB = cmdServer.Flag.Int("volume.concurrentDownloadLimitMB", 64, "limit total concurrent download size")
+	serverOptions.v.concurrentUploadLimitMB = cmdServer.Flag.Int("volume.concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited")
+	serverOptions.v.concurrentDownloadLimitMB = cmdServer.Flag.Int("volume.concurrentDownloadLimitMB", 0, "limit total concurrent download size, 0 means unlimited")
 	serverOptions.v.publicUrl = cmdServer.Flag.String("volume.publicUrl", "", "publicly accessible address")
 	serverOptions.v.preStopSeconds = cmdServer.Flag.Int("volume.preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
 	serverOptions.v.pprof = cmdServer.Flag.Bool("volume.pprof", false, "enable pprof http handlers. precludes -memprofile and -cpuprofile")
@@ -171,7 +171,7 @@ func init() {
 	s3Options.localSocket = cmdServer.Flag.String("s3.localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
 	s3Options.bindIp = cmdServer.Flag.String("s3.ip.bind", "", "ip address to bind to. If empty, default to same as -ip.bind option.")
 	s3Options.idleTimeout = cmdServer.Flag.Int("s3.idleTimeout", 120, "connection idle seconds")
-	s3Options.concurrentUploadLimitMB = cmdServer.Flag.Int("s3.concurrentUploadLimitMB", 128, "limit total concurrent upload size for S3")
+	s3Options.concurrentUploadLimitMB = cmdServer.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size for S3, 0 means unlimited")
 	s3Options.concurrentFileUploadLimit = cmdServer.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited")

 	sftpOptions.port = cmdServer.Flag.Int("sftp.port", 2022, "SFTP server listen port")
@@ -100,8 +100,8 @@ func init() {
 	v.maintenanceMBPerSecond = cmdVolume.Flag.Int("maintenanceMBps", 0, "limit maintenance (replication / balance) IO rate in MB/s. Unset is 0, no limitation.")
 	v.fileSizeLimitMB = cmdVolume.Flag.Int("fileSizeLimitMB", 256, "limit file size to avoid out of memory")
 	v.ldbTimeout = cmdVolume.Flag.Int64("index.leveldbTimeout", 0, "alive time for leveldb (default to 0). If leveldb of volume is not accessed in ldbTimeout hours, it will be off loaded to reduce opened files and memory consumption.")
-	v.concurrentUploadLimitMB = cmdVolume.Flag.Int("concurrentUploadLimitMB", 256, "limit total concurrent upload size")
-	v.concurrentDownloadLimitMB = cmdVolume.Flag.Int("concurrentDownloadLimitMB", 256, "limit total concurrent download size")
+	v.concurrentUploadLimitMB = cmdVolume.Flag.Int("concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited")
+	v.concurrentDownloadLimitMB = cmdVolume.Flag.Int("concurrentDownloadLimitMB", 0, "limit total concurrent download size, 0 means unlimited")
 	v.pprof = cmdVolume.Flag.Bool("pprof", false, "enable pprof http handlers. precludes -memprofile and -cpuprofile")
 	v.metricsHttpPort = cmdVolume.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 	v.metricsHttpIp = cmdVolume.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
@@ -19,6 +19,9 @@ import (
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance"
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding"
 	_ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
+	// TODO: Implement additional task packages (add to default capabilities when ready):
+	// _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/remote" - for uploading volumes to remote/cloud storage
+	// _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/replication" - for fixing replication issues and maintaining data consistency
 )

 var cmdWorker = &Command{
@@ -41,7 +44,7 @@ Examples:

 var (
 	workerAdminServer         = cmdWorker.Flag.String("admin", "localhost:23646", "admin server address")
-	workerCapabilities        = cmdWorker.Flag.String("capabilities", "vacuum,ec,remote,replication,balance", "comma-separated list of task types this worker can handle")
+	workerCapabilities        = cmdWorker.Flag.String("capabilities", "vacuum,ec,balance", "comma-separated list of task types this worker can handle")
 	workerMaxConcurrent       = cmdWorker.Flag.Int("maxConcurrent", 2, "maximum number of concurrent tasks")
 	workerHeartbeatInterval   = cmdWorker.Flag.Duration("heartbeat", 30*time.Second, "heartbeat interval")
 	workerTaskRequestInterval = cmdWorker.Flag.Duration("taskInterval", 5*time.Second, "task request interval")
@@ -152,9 +152,21 @@ func (f *Filer) logFlushFunc(logBuffer *log_buffer.LogBuffer, startTime, stopTim
 }

 var (
-	VolumeNotFoundPattern = regexp.MustCompile(`volume \d+? not found`)
+	volumeNotFoundPattern = regexp.MustCompile(`volume \d+? not found`)
+	chunkNotFoundPattern  = regexp.MustCompile(`(urls not found|File Not Found)`)
 )

+// isChunkNotFoundError checks if the error indicates that a volume or chunk
+// has been deleted and is no longer available. These errors can be skipped
+// when reading persisted log files since the data is unrecoverable.
+func isChunkNotFoundError(err error) bool {
+	if err == nil {
+		return false
+	}
+	errMsg := err.Error()
+	return volumeNotFoundPattern.MatchString(errMsg) || chunkNotFoundPattern.MatchString(errMsg)
+}
+
 func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) {

 	visitor, visitErr := f.collectPersistedLogBuffer(startPosition, stopTsNs)
@@ -265,11 +265,12 @@ func (c *LogFileEntryCollector) collectMore(v *OrderedLogVisitor) (err error) {
 // ----------

 type LogFileQueueIterator struct {
-	q                   *util.Queue[*LogFileEntry]
-	masterClient        *wdclient.MasterClient
-	startTsNs           int64
-	stopTsNs            int64
-	currentFileIterator *LogFileIterator
+	q              *util.Queue[*LogFileEntry]
+	masterClient   *wdclient.MasterClient
+	startTsNs      int64
+	stopTsNs       int64
+	pendingEntries []*filer_pb.LogEntry
+	pendingIndex   int
 }

 func newLogFileQueueIterator(masterClient *wdclient.MasterClient, q *util.Queue[*LogFileEntry], startTsNs, stopTsNs int64) *LogFileQueueIterator {
@@ -284,13 +285,17 @@ func newLogFileQueueIterator(masterClient *wdclient.MasterClient, q *util.Queue[
 // getNext will return io.EOF when done
 func (iter *LogFileQueueIterator) getNext(v *OrderedLogVisitor) (logEntry *filer_pb.LogEntry, err error) {
 	for {
-		if iter.currentFileIterator != nil {
-			logEntry, err = iter.currentFileIterator.getNext()
-			if err != io.EOF {
-				return
-			}
+		// return pending entries first
+		if iter.pendingIndex < len(iter.pendingEntries) {
+			logEntry = iter.pendingEntries[iter.pendingIndex]
+			iter.pendingIndex++
+			return logEntry, nil
 		}
-		// now either iter.currentFileIterator is nil or err is io.EOF
+		// reset for next file
+		iter.pendingEntries = nil
+		iter.pendingIndex = 0
+
+		// read entries from next file
 		if iter.q.Len() == 0 {
 			return nil, io.EOF
 		}
@@ -313,7 +318,38 @@ func (iter *LogFileQueueIterator) getNext(v *OrderedLogVisitor) (logEntry *filer
 		if next != nil && next.TsNs <= iter.startTsNs {
 			continue
 		}
-		iter.currentFileIterator = newLogFileIterator(iter.masterClient, t.FileEntry, iter.startTsNs, iter.stopTsNs)
+
+		// read all entries from this file
+		iter.pendingEntries, err = iter.readFileEntries(t.FileEntry)
+		if err != nil {
+			return nil, err
+		}
+	}
+}
+
+// readFileEntries reads all log entries from a single file
+func (iter *LogFileQueueIterator) readFileEntries(fileEntry *Entry) (entries []*filer_pb.LogEntry, err error) {
+	fileIterator := newLogFileIterator(iter.masterClient, fileEntry, iter.startTsNs, iter.stopTsNs)
+	defer func() {
+		if closeErr := fileIterator.Close(); closeErr != nil && err == nil {
+			err = closeErr
+		}
+	}()
+
+	for {
+		logEntry, err := fileIterator.getNext()
+		if err == io.EOF {
+			return entries, nil
+		}
+		if err != nil {
+			if isChunkNotFoundError(err) {
+				// Volume or chunk was deleted, skip the rest of this log file
+				glog.Warningf("skipping rest of %s: %v", fileIterator.filePath, err)
+				return entries, nil
+			}
+			return nil, err
+		}
+		entries = append(entries, logEntry)
 	}
 }

@@ -324,6 +360,7 @@ type LogFileIterator struct {
 	sizeBuf   []byte
 	startTsNs int64
 	stopTsNs  int64
+	filePath  string
 }

 func newLogFileIterator(masterClient *wdclient.MasterClient, fileEntry *Entry, startTsNs, stopTsNs int64) *LogFileIterator {
@@ -332,9 +369,17 @@ func newLogFileIterator(masterClient *wdclient.MasterClient, fileEntry *Entry, s
 		sizeBuf:   make([]byte, 4),
 		startTsNs: startTsNs,
 		stopTsNs:  stopTsNs,
+		filePath:  string(fileEntry.FullPath),
 	}
 }

+func (iter *LogFileIterator) Close() error {
+	if r, ok := iter.r.(io.Closer); ok {
+		return r.Close()
+	}
+	return nil
+}
+
 // getNext will return io.EOF when done
 func (iter *LogFileIterator) getNext() (logEntry *filer_pb.LogEntry, err error) {
 	var n int
@@ -22,6 +22,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"sync"
 	"time"

 	"github.com/apple/foundationdb/bindings/go/src/fdb"
@@ -37,15 +38,146 @@ import (
 const (
 	// FoundationDB transaction size limit is 10MB
 	FDB_TRANSACTION_SIZE_LIMIT = 10 * 1024 * 1024
+	// Safe limit for batch size (leave margin for FDB overhead)
+	FDB_BATCH_SIZE_LIMIT = 8 * 1024 * 1024
 	// Maximum number of entries to return in a single directory listing
 	// Large batches can cause transaction timeouts and increase memory pressure
 	MAX_DIRECTORY_LIST_LIMIT = 1000
+
+	// Write batching defaults
+	DEFAULT_BATCH_SIZE     = 100
+	DEFAULT_BATCH_INTERVAL = 5 * time.Millisecond
 )

 func init() {
 	filer.Stores = append(filer.Stores, &FoundationDBStore{})
 }

+// writeOp represents a pending write operation
+type writeOp struct {
+	key   fdb.Key
+	value []byte // nil for delete
+	done  chan error
+}
+
+// opSize returns the approximate size of an operation in bytes
+func (op *writeOp) size() int {
+	return len(op.key) + len(op.value)
+}
+
+// writeBatcher batches multiple writes into single transactions
+type writeBatcher struct {
+	store    *FoundationDBStore
+	ops      chan *writeOp
+	stop     chan struct{}
+	wg       sync.WaitGroup
+	size     int
+	interval time.Duration
+}
+
+func newWriteBatcher(store *FoundationDBStore, size int, interval time.Duration) *writeBatcher {
+	b := &writeBatcher{
+		store:    store,
+		ops:      make(chan *writeOp, size*10),
+		stop:     make(chan struct{}),
+		size:     size,
+		interval: interval,
+	}
+	b.wg.Add(1)
+	go b.run()
+	return b
+}
+
+func (b *writeBatcher) run() {
+	defer b.wg.Done()
+	batch := make([]*writeOp, 0, b.size)
+	batchBytes := 0 // Track cumulative size of batch
+	timer := time.NewTimer(b.interval)
+	defer timer.Stop()
+
+	flush := func() {
+		if len(batch) == 0 {
+			return
+		}
+		_, err := b.store.database.Transact(func(tr fdb.Transaction) (interface{}, error) {
+			for _, op := range batch {
+				if op.value != nil {
+					tr.Set(op.key, op.value)
+				} else {
+					tr.Clear(op.key)
+				}
+			}
+			return nil, nil
+		})
+		for _, op := range batch {
+			if op.done != nil {
+				op.done <- err
+				close(op.done)
+			}
+		}
+		batch = batch[:0]
+		batchBytes = 0
+	}
+
+	resetTimer := func() {
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+		timer.Reset(b.interval)
+	}
+
+	for {
+		select {
+		case op := <-b.ops:
+			batch = append(batch, op)
+			batchBytes += op.size()
+			// Flush when batch count or size limit is reached
+			if len(batch) >= b.size || batchBytes >= FDB_BATCH_SIZE_LIMIT {
+				flush()
+				resetTimer()
+			}
+		case <-timer.C:
+			flush()
+			// Timer already fired, safe to reset directly
+			timer.Reset(b.interval)
+		case <-b.stop:
+			for {
+				select {
+				case op := <-b.ops:
+					batch = append(batch, op)
+				default:
+					flush()
+					return
+				}
+			}
+		}
+	}
+}
+
+func (b *writeBatcher) submit(key fdb.Key, value []byte, wait bool) error {
+	op := &writeOp{key: key, value: value}
+	if wait {
+		op.done = make(chan error, 1)
+	}
+	select {
+	case b.ops <- op:
+		if wait {
+			return <-op.done
+		}
+		return nil
+	case <-b.stop:
+		return fmt.Errorf("batcher stopped")
+	}
+}
+
+func (b *writeBatcher) shutdown() {
+	close(b.stop)
+	b.wg.Wait()
+}
+
 type FoundationDBStore struct {
 	database        fdb.Database
 	seaweedfsDir    directory.DirectorySubspace
@@ -53,6 +185,10 @@ type FoundationDBStore struct {
 	directoryPrefix string
 	timeout         time.Duration
 	maxRetryDelay   time.Duration
+	// Write batching
+	batcher       *writeBatcher
+	batchSize     int
+	batchInterval time.Duration
 }

 // Context key type for storing transactions
@@ -89,6 +225,8 @@ func (store *FoundationDBStore) Initialize(configuration util.Configuration, pre
 	configuration.SetDefault(prefix+"timeout", "5s")
 	configuration.SetDefault(prefix+"max_retry_delay", "1s")
 	configuration.SetDefault(prefix+"directory_prefix", "seaweedfs")
+	configuration.SetDefault(prefix+"batch_size", DEFAULT_BATCH_SIZE)
+	configuration.SetDefault(prefix+"batch_interval", DEFAULT_BATCH_INTERVAL.String())

 	clusterFile := configuration.GetString(prefix + "cluster_file")
 	apiVersion := configuration.GetInt(prefix + "api_version")
@@ -108,6 +246,18 @@ func (store *FoundationDBStore) Initialize(configuration util.Configuration, pre
 		return fmt.Errorf("invalid max_retry_delay duration %s: %w", maxRetryDelayStr, err)
 	}

+	// Parse batch configuration
+	store.batchSize = configuration.GetInt(prefix + "batch_size")
+	if store.batchSize <= 0 {
+		store.batchSize = DEFAULT_BATCH_SIZE
+	}
+	batchIntervalStr := configuration.GetString(prefix + "batch_interval")
+	store.batchInterval, err = time.ParseDuration(batchIntervalStr)
+	if err != nil {
+		glog.Warningf("invalid %sbatch_interval duration %q, using default %v: %v", prefix, batchIntervalStr, DEFAULT_BATCH_INTERVAL, err)
+		store.batchInterval = DEFAULT_BATCH_INTERVAL
+	}
+
 	return store.initialize(clusterFile, apiVersion)
 }

@@ -138,6 +288,11 @@ func (store *FoundationDBStore) initialize(clusterFile string, apiVersion int) e
 		return fmt.Errorf("failed to create/open kv directory: %w", err)
 	}

+	// Start write batcher for improved throughput
+	store.batcher = newWriteBatcher(store, store.batchSize, store.batchInterval)
+	glog.V(0).Infof("FoundationDB: write batching enabled (batch_size=%d, batch_interval=%v)",
+		store.batchSize, store.batchInterval)
+
 	glog.V(0).Infof("FoundationDB store initialized successfully with directory prefix: %s", store.directoryPrefix)
 	return nil
 }
@@ -215,7 +370,12 @@ func (store *FoundationDBStore) UpdateEntry(ctx context.Context, entry *filer.En
 		return nil
 	}

-	// Execute in a new transaction if not in an existing one
+	// Use write batcher for better throughput
+	if store.batcher != nil {
+		return store.batcher.submit(key, value, true)
+	}
+
+	// Fallback: execute in a new transaction
 	_, err = store.database.Transact(func(tr fdb.Transaction) (interface{}, error) {
 		tr.Set(key, value)
 		return nil, nil
@@ -276,7 +436,12 @@ func (store *FoundationDBStore) DeleteEntry(ctx context.Context, fullpath util.F
 		return nil
 	}

-	// Execute in a new transaction if not in an existing one
+	// Use write batcher for better throughput (nil value = delete)
+	if store.batcher != nil {
+		return store.batcher.submit(key, nil, true)
+	}
+
+	// Fallback: execute in a new transaction
 	_, err := store.database.Transact(func(tr fdb.Transaction) (interface{}, error) {
 		tr.Clear(key)
 		return nil, nil
@@ -556,6 +721,11 @@ func (store *FoundationDBStore) KvDelete(ctx context.Context, key []byte) error
 }

 func (store *FoundationDBStore) Shutdown() {
+	// Stop write batcher
+	if store.batcher != nil {
+		store.batcher.shutdown()
+		store.batcher = nil
+	}
 	// FoundationDB doesn't have an explicit close method for Database
 	glog.V(0).Infof("FoundationDB store shutdown")
 }
@@ -107,6 +107,39 @@ func (store *LevelDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry)
 	return store.InsertEntry(ctx, entry)
 }

+// BatchInsertEntries inserts multiple entries in a single LevelDB batch write.
+// This is more efficient than inserting entries one by one as it reduces
+// the number of write operations and syncs to disk.
+func (store *LevelDBStore) BatchInsertEntries(ctx context.Context, entries []*filer.Entry) error {
+	if len(entries) == 0 {
+		return nil
+	}
+
+	batch := new(leveldb.Batch)
+
+	for _, entry := range entries {
+		key := genKey(entry.DirAndName())
+
+		value, err := entry.EncodeAttributesAndChunks()
+		if err != nil {
+			return fmt.Errorf("encoding %s %+v: %w", entry.FullPath, entry.Attr, err)
+		}
+
+		if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
+			value = weed_util.MaybeGzipData(value)
+		}
+
+		batch.Put(key, value)
+	}
+
+	err := store.db.Write(batch, nil)
+	if err != nil {
+		return fmt.Errorf("batch write: %w", err)
+	}
+
+	return nil
+}
+
 func (store *LevelDBStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
 	key := genKey(fullpath.DirAndName())

@@ -245,6 +245,7 @@ type ChunkStreamReader struct {

 var _ = io.ReadSeeker(&ChunkStreamReader{})
 var _ = io.ReaderAt(&ChunkStreamReader{})
+var _ = io.Closer(&ChunkStreamReader{})

 func doNewChunkStreamReader(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk) *ChunkStreamReader {

@@ -403,8 +404,13 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 	return nil
 }

-func (c *ChunkStreamReader) Close() {
-	// TODO try to release and reuse buffer
+func (c *ChunkStreamReader) Close() error {
+	c.bufferLock.Lock()
+	defer c.bufferLock.Unlock()
+	c.buffer = nil
+	c.head = nil
+	c.chunkView = nil
+	return nil
 }

 func VolumeId(fileId string) string {
@@ -1,11 +1,12 @@
 package mount

 import (
+	"sync"
+	"time"
+
 	"github.com/hanwen/go-fuse/v2/fuse"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/util"
-	"sync"
-	"time"
 )

 type InodeToPath struct {
@@ -41,7 +42,6 @@ func (ie *InodeEntry) removeOnePath(p util.FullPath) bool {
 		ie.paths[x] = ie.paths[x+1]
 	}
 	ie.paths = ie.paths[0 : len(ie.paths)-1]
-	ie.nlookup--
 	return true
 }

@@ -160,6 +160,10 @@ func (i *InodeToPath) MarkChildrenCached(fullpath util.FullPath) {
 		return
 	}
 	path, found := i.inode2path[inode]
+	if !found {
+		glog.Warningf("MarkChildrenCached inode %d not found in inode2path for %v", inode, fullpath)
+		return
+	}
 	path.isChildrenCached = true
 	if i.cacheMetaTtlSec > 0 {
 		path.cachedExpiresTime = time.Now().Add(i.cacheMetaTtlSec)
@@ -230,9 +234,6 @@ func (i *InodeToPath) removePathFromInode2Path(inode uint64, path util.FullPath)
 	if !ie.removeOnePath(path) {
 		return
 	}
-	if len(ie.paths) == 0 {
-		delete(i.inode2path, inode)
-	}
 }

 func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (sourceInode, targetInode uint64) {
@@ -259,9 +260,6 @@ func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (sourceInod
 			}
 		}
 		entry.isChildrenCached = false
-		if !targetFound {
-			entry.nlookup++
-		}
 	} else {
 		glog.Errorf("MovePath %s to %s: sourceInode %d not found", sourcePath, targetPath, sourceInode)
 	}
@@ -269,24 +267,39 @@ func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (sourceInod
 }

 func (i *InodeToPath) Forget(inode, nlookup uint64, onForgetDir func(dir util.FullPath)) {
+	var dirPaths []util.FullPath
+	callOnForgetDir := false
+
 	i.Lock()
 	path, found := i.inode2path[inode]
 	if found {
-		path.nlookup -= nlookup
-		if path.nlookup <= 0 {
+		if nlookup > path.nlookup {
+			glog.Errorf("kernel forget over-decrement: inode %d paths %v current %d forget %d", inode, path.paths, path.nlookup, nlookup)
+			path.nlookup = 0
+		} else {
+			path.nlookup -= nlookup
+		}
+		glog.V(4).Infof("kernel forget: inode %d paths %v nlookup %d", inode, path.paths, path.nlookup)
+		if path.nlookup == 0 {
+			if path.isDirectory && onForgetDir != nil {
+				dirPaths = append([]util.FullPath(nil), path.paths...)
+				callOnForgetDir = true
+			}
 			for _, p := range path.paths {
 				delete(i.path2inode, p)
 			}
 			delete(i.inode2path, inode)
+		} else {
+			glog.V(4).Infof("kernel forget but nlookup not zero: inode %d paths %v nlookup %d", inode, path.paths, path.nlookup)
 		}
+	} else {
+		glog.Warningf("kernel forget but inode not found: inode %d", inode)
 	}
 	i.Unlock()
-	if found {
-		if path.isDirectory && path.nlookup <= 0 && onForgetDir != nil {
-			path.isChildrenCached = false
-			for _, p := range path.paths {
-				onForgetDir(p)
-			}
+
+	if callOnForgetDir {
+		for _, p := range dirPaths {
+			onForgetDir(p)
 		}
 	}
 }
@@ -6,6 +6,8 @@ import (
 	"sync"
 	"time"

+	"golang.org/x/sync/singleflight"
+
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/filer/leveldb"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -17,20 +19,24 @@ import (
 // e.g. fill fileId field for chunks

 type MetaCache struct {
-	root       util.FullPath
-	localStore filer.VirtualFilerStore
+	root        util.FullPath
+	localStore  filer.VirtualFilerStore
+	leveldbStore *leveldb.LevelDBStore // direct reference for batch operations
 	sync.RWMutex
 	uidGidMapper   *UidGidMapper
 	markCachedFn   func(fullpath util.FullPath)
 	isCachedFn     func(fullpath util.FullPath) bool
 	invalidateFunc func(fullpath util.FullPath, entry *filer_pb.Entry)
+	visitGroup     singleflight.Group // deduplicates concurrent EnsureVisited calls for the same path
 }

 func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath,
 	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(util.FullPath, *filer_pb.Entry)) *MetaCache {
+	leveldbStore, virtualStore := openMetaStore(dbFolder)
 	return &MetaCache{
 		root:         root,
-		localStore:   openMetaStore(dbFolder),
+		localStore:   virtualStore,
+		leveldbStore: leveldbStore,
 		markCachedFn: markCachedFn,
 		isCachedFn:   isCachedFn,
 		uidGidMapper: uidGidMapper,
@@ -40,7 +46,7 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat
 	}
 }

-func openMetaStore(dbFolder string) filer.VirtualFilerStore {
+func openMetaStore(dbFolder string) (*leveldb.LevelDBStore, filer.VirtualFilerStore) {

 	os.RemoveAll(dbFolder)
 	os.MkdirAll(dbFolder, 0755)
@@ -54,7 +60,7 @@ func openMetaStore(dbFolder string) filer.VirtualFilerStore {
 		glog.Fatalf("Failed to initialize metadata cache store for %s: %+v", store.GetName(), err)
 	}

-	return filer.NewFilerStoreWrapper(store)
+	return store, filer.NewFilerStoreWrapper(store)

 }

@@ -68,6 +74,12 @@ func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) erro
 	return mc.localStore.InsertEntry(ctx, entry)
 }

+// doBatchInsertEntries inserts multiple entries using LevelDB's batch write.
+// This is more efficient than inserting entries one by one.
+func (mc *MetaCache) doBatchInsertEntries(ctx context.Context, entries []*filer.Entry) error {
+	return mc.leveldbStore.BatchInsertEntries(ctx, entries)
+}
+
 func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
 	mc.Lock()
 	defer mc.Unlock()
@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"

+	"golang.org/x/sync/errgroup"
+
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -11,22 +13,18 @@ import (
 )

 func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath) error {
-
+	// Collect all uncached paths from target directory up to root
+	var uncachedPaths []util.FullPath
 	currentPath := dirPath

 	for {
-
-		// the directory children are already cached
-		// so no need for this and upper directories
+		// If this path is cached, all ancestors are also cached
 		if mc.isCachedFn(currentPath) {
-			return nil
+			break
 		}
+		uncachedPaths = append(uncachedPaths, currentPath)

-		if err := doEnsureVisited(mc, client, currentPath); err != nil {
-			return err
-		}
-
-		// continue to parent directory
+		// Continue to parent directory
 		if currentPath != mc.root {
 			parent, _ := currentPath.DirAndName()
 			currentPath = util.FullPath(parent)
@@ -35,33 +33,82 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full
 		}
 	}

-	return nil
+	if len(uncachedPaths) == 0 {
+		return nil
+	}

+	// Fetch all uncached directories in parallel with context for cancellation
+	// If one fetch fails, cancel the others to avoid unnecessary work
+	g, ctx := errgroup.WithContext(context.Background())
+	for _, p := range uncachedPaths {
+		path := p // capture for closure
+		g.Go(func() error {
+			return doEnsureVisited(ctx, mc, client, path)
+		})
+	}
+	return g.Wait()
 }

-func doEnsureVisited(mc *MetaCache, client filer_pb.FilerClient, path util.FullPath) error {
+// batchInsertSize is the number of entries to accumulate before flushing to LevelDB.
+// 100 provides a balance between memory usage (~100 Entry pointers) and write efficiency
+// (fewer disk syncs). Larger values reduce I/O overhead but increase memory and latency.
+const batchInsertSize = 100

-	glog.V(4).Infof("ReadDirAllEntries %s ...", path)
+func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, path util.FullPath) error {
+	// Use singleflight to deduplicate concurrent requests for the same path
+	_, err, _ := mc.visitGroup.Do(string(path), func() (interface{}, error) {
+		// Check for cancellation before starting
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}

-	err := util.Retry("ReadDirAllEntries", func() error {
-		return filer_pb.ReadDirAllEntries(context.Background(), client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
-			entry := filer.FromPbEntry(string(path), pbEntry)
-			if IsHiddenSystemEntry(string(path), entry.Name()) {
+		// Double-check if already cached (another goroutine may have completed)
+		if mc.isCachedFn(path) {
+			return nil, nil
+		}
+
+		glog.V(4).Infof("ReadDirAllEntries %s ...", path)
+
+		// Collect entries in batches for efficient LevelDB writes
+		var batch []*filer.Entry
+
+		fetchErr := util.Retry("ReadDirAllEntries", func() error {
+			batch = nil // Reset batch on retry, allow GC of previous entries
+			return filer_pb.ReadDirAllEntries(ctx, client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
+				entry := filer.FromPbEntry(string(path), pbEntry)
+				if IsHiddenSystemEntry(string(path), entry.Name()) {
+					return nil
+				}
+
+				batch = append(batch, entry)
+
+				// Flush batch when it reaches the threshold
+				// Don't rely on isLast here - hidden entries may cause early return
+				if len(batch) >= batchInsertSize {
+					// No lock needed - LevelDB Write() is thread-safe
+					if err := mc.doBatchInsertEntries(ctx, batch); err != nil {
+						return fmt.Errorf("batch insert for %s: %w", path, err)
+					}
+					// Create new slice to allow GC of flushed entries
+					batch = make([]*filer.Entry, 0, batchInsertSize)
+				}
 				return nil
-			}
-			if err := mc.doInsertEntry(context.Background(), entry); err != nil {
-				glog.V(0).Infof("read %s: %v", entry.FullPath, err)
-				return err
-			}
-			return nil
+			})
 		})
-	})

-	if err != nil {
-		err = fmt.Errorf("list %s: %v", path, err)
-	} else {
+		if fetchErr != nil {
+			return nil, fmt.Errorf("list %s: %w", path, fetchErr)
+		}
+
+		// Flush any remaining entries in the batch
+		if len(batch) > 0 {
+			if err := mc.doBatchInsertEntries(ctx, batch); err != nil {
+				return nil, fmt.Errorf("batch insert remaining for %s: %w", path, err)
+			}
+		}
 		mc.markCachedFn(path)
-	}
+		return nil, nil
+	})
 	return err
 }

@@ -64,6 +64,10 @@ type Option struct {
 	Cipher             bool   // whether encrypt data on volume server
 	UidGidMapper       *meta_cache.UidGidMapper

+	// Periodic metadata flush interval in seconds (0 to disable)
+	// This protects chunks from being purged by volume.fsck for long-running writes
+	MetadataFlushSeconds int
+
 	// RDMA acceleration options
 	RdmaEnabled       bool
 	RdmaSidecarAddr   string
@@ -214,6 +218,7 @@ func (wfs *WFS) StartBackgroundTasks() error {
 	startTime := time.Now()
 	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano(), follower)
 	go wfs.loopCheckQuota()
+	go wfs.loopFlushDirtyMetadata()

 	return nil
 }
@@ -2,7 +2,6 @@ package mount

 import (
 	"context"
-	"math"
 	"sync"

 	"github.com/hanwen/go-fuse/v2/fuse"
@@ -16,6 +15,7 @@ type DirectoryHandleId uint64

 const (
 	directoryStreamBaseOffset = 2 // . & ..
+	batchSize                 = 1000
 )

 // DirectoryHandle represents an open directory handle.
@@ -159,27 +159,30 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPl
 		}
 	}

-	isEarlyTerminated := false
 	dirPath, code := wfs.inodeToPath.GetPath(input.NodeId)
 	if code != fuse.OK {
 		return code
 	}

 	var dirEntry fuse.DirEntry
-	processEachEntryFn := func(entry *filer.Entry) bool {
+
+	// index is the position in entryStream, used to calculate the offset for next readdir
+	processEachEntryFn := func(entry *filer.Entry, index int64) bool {
 		dirEntry.Name = entry.Name()
 		dirEntry.Mode = toSyscallMode(entry.Mode)
-		inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Crtime.Unix(), entry.IsDirectory(), len(entry.HardLinkId) > 0, entry.Inode, isPlusMode)
+		inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Crtime.Unix(), entry.IsDirectory(), len(entry.HardLinkId) > 0, entry.Inode, false)
 		dirEntry.Ino = inode
+
+		// Set Off to the next offset so client can resume from correct position
+		dirEntry.Off = dh.entryStreamOffset + uint64(index) + 1
+
 		if !isPlusMode {
 			if !out.AddDirEntry(dirEntry) {
-				isEarlyTerminated = true
 				return false
 			}
 		} else {
 			entryOut := out.AddDirLookupEntry(dirEntry)
 			if entryOut == nil {
-				isEarlyTerminated = true
 				return false
 			}
 			if fh, found := wfs.fhMap.FindFileHandle(inode); found {
@@ -187,6 +190,7 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPl
 				entry = filer.FromPbEntry(string(dirPath), fh.GetEntry().GetEntry())
 			}
 			wfs.outputFilerEntry(entryOut, inode, entry)
+			wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Crtime.Unix(), entry.IsDirectory(), len(entry.HardLinkId) > 0, entry.Inode, true)
 		}
 		return true
 	}
@@ -194,59 +198,90 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPl
 	if input.Offset < directoryStreamBaseOffset {
 		if !isPlusMode {
 			if input.Offset == 0 {
-				out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "."})
+				out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".", Off: 1})
 			}
-			out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".."})
+			out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "..", Off: 2})
 		} else {
 			if input.Offset == 0 {
-				out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "."})
+				out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".", Off: 1})
 			}
-			out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".."})
+			out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "..", Off: 2})
 		}
 		input.Offset = directoryStreamBaseOffset
 	}

 	var lastEntryName string
+
+	// Read from cache first, then load next batch if needed
 	if input.Offset >= dh.entryStreamOffset {
+		// Handle case: new handle with non-zero offset but empty cache
+		// This happens when NFS-Ganesha opens multiple directory handles
+		if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset {
+			skipCount := int64(input.Offset - dh.entryStreamOffset)
+
+			if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath); err != nil {
+				glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
+				return fuse.EIO
+			}
+
+			// Load entries from beginning to fill cache up to the requested offset
+			loadErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) {
+				dh.entryStream = append(dh.entryStream, entry)
+				return true, nil
+			})
+			if loadErr != nil {
+				glog.Errorf("list meta cache: %v", loadErr)
+				return fuse.EIO
+			}
+		}
+
 		if input.Offset > dh.entryStreamOffset {
 			entryPreviousIndex := (input.Offset - dh.entryStreamOffset) - 1
 			if uint64(len(dh.entryStream)) > entryPreviousIndex {
 				lastEntryName = dh.entryStream[entryPreviousIndex].Name()
-				dh.entryStream = dh.entryStream[entryPreviousIndex:]
-				dh.entryStreamOffset = input.Offset - 1
 			}
 		}
-		entryCurrentIndex := input.Offset - dh.entryStreamOffset
-		for uint64(len(dh.entryStream)) > entryCurrentIndex {
+
+		entryCurrentIndex := int64(input.Offset - dh.entryStreamOffset)
+		for int64(len(dh.entryStream)) > entryCurrentIndex {
 			entry := dh.entryStream[entryCurrentIndex]
-			if processEachEntryFn(entry) {
+			if processEachEntryFn(entry, entryCurrentIndex) {
 				lastEntryName = entry.Name()
 				entryCurrentIndex++
 			} else {
-				// early terminated
 				return fuse.OK
 			}
 		}
-	}

-	if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath); err != nil {
-		glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
-		return fuse.EIO
-	}
-	listErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, lastEntryName, false, int64(math.MaxInt32), func(entry *filer.Entry) (bool, error) {
-		dh.entryStream = append(dh.entryStream, entry)
-		if !processEachEntryFn(entry) {
-			return false, nil
+		// Cache exhausted, load next batch
+		if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath); err != nil {
+			glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
+			return fuse.EIO
 		}
-		return true, nil
-	})
-	if listErr != nil {
-		glog.Errorf("list meta cache: %v", listErr)
-		return fuse.EIO
-	}

-	if !isEarlyTerminated {
-		dh.isFinished = true
+		// Batch loading: fetch batchSize entries starting from lastEntryName
+		loadedCount := 0
+		bufferFull := false
+		loadErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) {
+			currentIndex := int64(len(dh.entryStream))
+			dh.entryStream = append(dh.entryStream, entry)
+			loadedCount++
+			if !processEachEntryFn(entry, currentIndex) {
+				bufferFull = true
+				return false, nil
+			}
+			return true, nil
+		})
+		if loadErr != nil {
+			glog.Errorf("list meta cache: %v", loadErr)
+			return fuse.EIO
+		}
+
+		// Mark finished only when loading completed normally (not buffer full)
+		// and we got fewer entries than requested
+		if !bufferFull && loadedCount < batchSize {
+			dh.isFinished = true
+		}
 	}

 	return fuse.OK
@@ -0,0 +1,168 @@
+package mount
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// loopFlushDirtyMetadata periodically flushes dirty file metadata to the filer.
+// This protects newly uploaded chunks from being purged by volume.fsck orphan cleanup
+// for files that remain open for extended periods without being closed.
+//
+// The problem: When a file is opened and written to continuously, chunks are uploaded
+// to volume servers but the file metadata (containing chunk references) is only saved
+// to the filer on file close or fsync. If volume.fsck runs during this window, it may
+// identify these chunks as orphans (since they're not referenced in filer metadata)
+// and purge them.
+//
+// This background task periodically flushes metadata for open files, ensuring chunk
+// references are visible to volume.fsck even before files are closed.
+func (wfs *WFS) loopFlushDirtyMetadata() {
+	if wfs.option.MetadataFlushSeconds <= 0 {
+		glog.V(0).Infof("periodic metadata flush disabled")
+		return
+	}
+
+	flushInterval := time.Duration(wfs.option.MetadataFlushSeconds) * time.Second
+	glog.V(0).Infof("periodic metadata flush enabled, interval: %v", flushInterval)
+
+	ticker := time.NewTicker(flushInterval)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		wfs.flushAllDirtyMetadata()
+	}
+}
+
+// flushAllDirtyMetadata iterates through all open file handles and flushes
+// metadata for files that have dirty metadata (chunks uploaded but not yet persisted).
+func (wfs *WFS) flushAllDirtyMetadata() {
+	// Collect file handles with dirty metadata under a read lock
+	var dirtyHandles []*FileHandle
+	wfs.fhMap.RLock()
+	for _, fh := range wfs.fhMap.inode2fh {
+		if fh.dirtyMetadata {
+			dirtyHandles = append(dirtyHandles, fh)
+		}
+	}
+	wfs.fhMap.RUnlock()
+
+	if len(dirtyHandles) == 0 {
+		return
+	}
+
+	glog.V(3).Infof("flushing metadata for %d open files", len(dirtyHandles))
+
+	// Process dirty handles in parallel with limited concurrency
+	var wg sync.WaitGroup
+	concurrency := wfs.option.ConcurrentWriters
+	if concurrency <= 0 {
+		concurrency = 16
+	}
+	sem := make(chan struct{}, concurrency)
+
+	for _, fh := range dirtyHandles {
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(handle *FileHandle) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			if err := wfs.flushFileMetadata(handle); err != nil {
+				glog.Warningf("failed to flush metadata for %s: %v", handle.FullPath(), err)
+			}
+		}(fh)
+	}
+	wg.Wait()
+}
+
+// flushFileMetadata flushes the current file metadata to the filer without
+// flushing dirty pages from memory. This updates chunk references in the filer
+// so volume.fsck can see them, while keeping data in the write buffer.
+func (wfs *WFS) flushFileMetadata(fh *FileHandle) error {
+	// Acquire exclusive lock on the file handle
+	fhActiveLock := fh.wfs.fhLockTable.AcquireLock("flushMetadata", fh.fh, util.ExclusiveLock)
+	defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
+
+	// Double-check dirty flag under lock
+	if !fh.dirtyMetadata {
+		return nil
+	}
+
+	fileFullPath := fh.FullPath()
+	dir, name := fileFullPath.DirAndName()
+
+	glog.V(4).Infof("flushFileMetadata %s fh %d", fileFullPath, fh.fh)
+
+	err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		entry := fh.GetEntry()
+		if entry == nil {
+			return nil
+		}
+		entry.Name = name
+
+		if entry.Attributes != nil {
+			entry.Attributes.Mtime = time.Now().Unix()
+		}
+
+		// Get current chunks - these include chunks that have been uploaded
+		// but not yet persisted to filer metadata
+		chunks := entry.GetChunks()
+		if len(chunks) == 0 {
+			return nil
+		}
+
+		// Separate manifest and non-manifest chunks
+		manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(chunks)
+
+		// Compact chunks to remove fully overlapped ones
+		compactedChunks, _ := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks)
+
+		// Try to create manifest chunks for large files
+		compactedChunks, manifestErr := filer.MaybeManifestize(wfs.saveDataAsChunk(fileFullPath), compactedChunks)
+		if manifestErr != nil {
+			glog.V(0).Infof("flushFileMetadata MaybeManifestize: %v", manifestErr)
+		}
+
+		entry.Chunks = append(compactedChunks, manifestChunks...)
+
+		request := &filer_pb.CreateEntryRequest{
+			Directory:                string(dir),
+			Entry:                    entry.GetEntry(),
+			Signatures:               []int32{wfs.signature},
+			SkipCheckParentDirectory: true,
+		}
+
+		wfs.mapPbIdFromLocalToFiler(request.Entry)
+		defer wfs.mapPbIdFromFilerToLocal(request.Entry)
+
+		if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
+			return err
+		}
+
+		// Update meta cache
+		if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry)); err != nil {
+			return fmt.Errorf("update meta cache for %s: %w", fileFullPath, err)
+		}
+
+		glog.V(3).Infof("flushed metadata for %s with %d chunks", fileFullPath, len(entry.GetChunks()))
+		return nil
+	})
+
+	if err != nil {
+		return err
+	}
+
+	// Note: We do NOT clear dirtyMetadata here because:
+	// 1. There may still be dirty pages in the write buffer
+	// 2. The file may receive more writes before close
+	// 3. dirtyMetadata will be cleared on the final flush when the file is closed
+
+	return nil
+}
@@ -4,12 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/seaweedfs/seaweedfs/weed/pb"
-	"google.golang.org/grpc"
 	"math/rand/v2"
 	"strings"
 	"time"

+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"google.golang.org/grpc"
+
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 )

@@ -101,7 +102,7 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 					GrpcPort:   int(loc.GrpcPort),
 				})
 			}
-			if vidLocations.Error != "" {
+			if vidLocations.Error == "" {
 				vc.Set(vidLocations.VolumeOrFileId, locations, 10*time.Minute)
 			}
 			ret[vidLocations.VolumeOrFileId] = &LookupResult{
@@ -582,8 +582,10 @@ func (iam *IdentityAccessManagement) authRequest(r *http.Request, action Action)
 		// - No policy or indeterminate → fall through to IAM checks
 		if iam.policyEngine != nil && bucket != "" {
 			principal := buildPrincipalARN(identity)
-			// Use context-aware policy evaluation to get the correct S3 action
-			allowed, evaluated, err := iam.policyEngine.EvaluatePolicyWithContext(bucket, object, string(action), principal, r)
+			// Phase 1: Evaluate bucket policy without object entry.
+			// Tag-based conditions (s3:ExistingObjectTag) are re-checked by handlers
+			// after fetching the entry, which is the Phase 2 check.
+			allowed, evaluated, err := iam.policyEngine.EvaluatePolicy(bucket, object, string(action), principal, r, nil)

 			if err != nil {
 				// SECURITY: Fail-close on policy evaluation errors
@@ -135,8 +135,70 @@ Standard AWS condition keys are supported:
 - `aws:UserAgent` - Client user agent
 - `s3:x-amz-acl` - Requested ACL
 - `s3:VersionId` - Object version ID
+- `s3:ExistingObjectTag/<tag-key>` - Value of an existing object tag (see example below)
 - And many more...

+### 5. Object Tag-Based Access Control
+
+You can control access based on object tags using `s3:ExistingObjectTag/<tag-key>`:
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": "*",
+      "Action": "s3:GetObject",
+      "Resource": "arn:aws:s3:::my-bucket/*",
+      "Condition": {
+        "StringEquals": {
+          "s3:ExistingObjectTag/status": ["public"]
+        }
+      }
+    }
+  ]
+}
+```
+
+This allows anonymous access only to objects that have a tag `status=public`.
+
+**Deny access to confidential objects:**
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": "*",
+      "Action": "s3:GetObject",
+      "Resource": "arn:aws:s3:::my-bucket/*"
+    },
+    {
+      "Effect": "Deny",
+      "Principal": "*",
+      "Action": "s3:GetObject",
+      "Resource": "arn:aws:s3:::my-bucket/*",
+      "Condition": {
+        "StringEquals": {
+          "s3:ExistingObjectTag/classification": ["confidential", "secret"]
+        }
+      }
+    }
+  ]
+}
+```
+
+**Supported Operations for Tag-Based Conditions:**
+
+Tag-based conditions (`s3:ExistingObjectTag/<key>`) are evaluated for the following operations:
+- `s3:GetObject` (GET object)
+- `s3:GetObjectVersion` (GET object with versionId)
+- `s3:HeadObject` (HEAD object)
+
+Note: For these conditions to be evaluated, the object must exist and the policy engine re-checks access after fetching the object metadata.
+
 ## Policy Evaluation

 ### Evaluation Order (AWS-Compatible)
@@ -270,10 +332,39 @@ go test -v -run TestPolicyValidation

 ## Compatibility

- ✅ **Full backward compatibility** with existing `identities.json`
- ✅ **AWS S3 API compatibility** for bucket policies
- ✅ **Standard condition operators** and keys
- ✅ **Proper evaluation precedence** (Deny > Allow > Default Deny)
- ✅ **Performance optimized** with caching and compiled patterns
+- Full backward compatibility with existing `identities.json`
+- AWS S3 API compatibility for bucket policies
+- Standard condition operators and keys
+- Proper evaluation precedence (Deny > Allow > Default Deny)
+- Performance optimized with caching and compiled patterns

-The policy engine provides a seamless upgrade path from SeaweedFS's existing simple IAM system to full AWS S3-compatible policies, giving you the best of both worlds: simplicity for basic use cases and power for complex enterprise scenarios. 
+The policy engine provides a seamless upgrade path from SeaweedFS's existing simple IAM system to full AWS S3-compatible policies, giving you the best of both worlds: simplicity for basic use cases and power for complex enterprise scenarios.
+
+## Feature Status
+
+### Implemented
+
+| Feature | Description |
+|---------|-------------|
+| Bucket Policies | Full AWS S3-compatible bucket policies |
+| Condition Operators | StringEquals, IpAddress, Bool, DateGreaterThan, etc. |
+| `aws:SourceIp` | IP-based access control with CIDR support |
+| `aws:SecureTransport` | Require HTTPS |
+| `aws:CurrentTime` | Time-based access control |
+| `s3:ExistingObjectTag/<key>` | Tag-based access control for existing objects |
+| Wildcard Patterns | Support for `*` and `?` in actions and resources |
+| Principal Matching | `*`, account IDs, and user ARNs |
+
+### Planned
+
+| Feature | GitHub Issue |
+|---------|--------------|
+| `s3:RequestObjectTag/<key>` | For tag conditions on PUT requests |
+| `s3:RequestObjectTagKeys` | Check which tag keys are in request |
+| `s3:x-amz-content-sha256` | Content hash condition |
+| `s3:x-amz-server-side-encryption` | SSE condition |
+| `s3:x-amz-storage-class` | Storage class condition |
+| Cross-account access | Access across different accounts |
+| VPC Endpoint policies | Network-level policies |
+
+For feature requests or to track progress, see the [GitHub Issues](https://github.com/seaweedfs/seaweedfs/issues). 
@@ -10,6 +10,7 @@ import (
 	"time"

 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )

 // LRUNode represents a node in the doubly-linked list for efficient LRU operations
@@ -705,8 +706,36 @@ func GetConditionEvaluator(operator string) (ConditionEvaluator, error) {
 	}
 }

+// ExistingObjectTagPrefix is the prefix for S3 policy condition keys
+const ExistingObjectTagPrefix = "s3:ExistingObjectTag/"
+
+// getConditionContextValue resolves the value(s) for a condition key.
+// For s3:ExistingObjectTag/<key> conditions, it looks up the tag in objectEntry.
+// For other condition keys, it looks up the value in contextValues.
+func getConditionContextValue(key string, contextValues map[string][]string, objectEntry map[string][]byte) []string {
+	if strings.HasPrefix(key, ExistingObjectTagPrefix) {
+		tagKey := key[len(ExistingObjectTagPrefix):]
+		if tagKey == "" {
+			return []string{} // Invalid: empty tag key
+		}
+		metadataKey := s3_constants.AmzObjectTaggingPrefix + tagKey
+		if objectEntry != nil {
+			if tagValue, exists := objectEntry[metadataKey]; exists {
+				return []string{string(tagValue)}
+			}
+		}
+		return []string{}
+	}
+
+	if vals, exists := contextValues[key]; exists {
+		return vals
+	}
+	return []string{}
+}
+
 // EvaluateConditions evaluates all conditions in a policy statement
-func EvaluateConditions(conditions PolicyConditions, contextValues map[string][]string) bool {
+// objectEntry is the object's metadata from entry.Extended (can be nil)
+func EvaluateConditions(conditions PolicyConditions, contextValues map[string][]string, objectEntry map[string][]byte) bool {
 	if len(conditions) == 0 {
 		return true // No conditions means always true
 	}
@@ -719,11 +748,7 @@ func EvaluateConditions(conditions PolicyConditions, contextValues map[string][]
 		}

 		for key, value := range conditionMap {
-			contextVals, exists := contextValues[key]
-			if !exists {
-				contextVals = []string{}
-			}
-
+			contextVals := getConditionContextValue(key, contextValues, objectEntry)
 			if !conditionEvaluator.Evaluate(value.Strings(), contextVals) {
 				return false // If any condition fails, the whole condition block fails
 			}
@@ -734,7 +759,8 @@ func EvaluateConditions(conditions PolicyConditions, contextValues map[string][]
 }

 // EvaluateConditionsLegacy evaluates conditions using the old interface{} format for backward compatibility
-func EvaluateConditionsLegacy(conditions map[string]interface{}, contextValues map[string][]string) bool {
+// objectEntry is the object's metadata from entry.Extended (can be nil)
+func EvaluateConditionsLegacy(conditions map[string]interface{}, contextValues map[string][]string, objectEntry map[string][]byte) bool {
 	if len(conditions) == 0 {
 		return true // No conditions means always true
 	}
@@ -753,11 +779,7 @@ func EvaluateConditionsLegacy(conditions map[string]interface{}, contextValues m
 		}

 		for key, value := range conditionMapTyped {
-			contextVals, exists := contextValues[key]
-			if !exists {
-				contextVals = []string{}
-			}
-
+			contextVals := getConditionContextValue(key, contextValues, objectEntry)
 			if !conditionEvaluator.Evaluate(value, contextVals) {
 				return false // If any condition fails, the whole condition block fails
 			}
@@ -91,6 +91,14 @@ func (engine *PolicyEngine) DeleteBucketPolicy(bucketName string) error {
 	return nil
 }

+// HasPolicyForBucket checks if a bucket has a policy configured
+func (engine *PolicyEngine) HasPolicyForBucket(bucketName string) bool {
+	engine.mutex.RLock()
+	defer engine.mutex.RUnlock()
+	_, exists := engine.contexts[bucketName]
+	return exists
+}
+
 // EvaluatePolicy evaluates a policy for the given arguments
 func (engine *PolicyEngine) EvaluatePolicy(bucketName string, args *PolicyEvaluationArgs) PolicyEvaluationResult {
 	engine.mutex.RLock()
@@ -154,7 +162,7 @@ func (engine *PolicyEngine) evaluateStatement(stmt *CompiledStatement, args *Pol

 	// Check conditions
 	if len(stmt.Statement.Condition) > 0 {
-		if !EvaluateConditions(stmt.Statement.Condition, args.Conditions) {
+		if !EvaluateConditions(stmt.Statement.Condition, args.Conditions, args.ObjectEntry) {
 			return false
 		}
 	}
@@ -201,10 +209,8 @@ func ExtractConditionValuesFromRequest(r *http.Request) map[string][]string {
 		values["aws:Referer"] = []string{referer}
 	}

-	// S3 object-level conditions
-	if r.Method == "GET" || r.Method == "HEAD" {
-		values["s3:ExistingObjectTag"] = extractObjectTags(r)
-	}
+	// Note: s3:ExistingObjectTag/<key> conditions are evaluated using objectEntry
+	// passed to EvaluatePolicy, not extracted from the request.

 	// S3 bucket-level conditions
 	if delimiter := r.URL.Query().Get("delimiter"); delimiter != "" {
@@ -243,13 +249,6 @@ func ExtractConditionValuesFromRequest(r *http.Request) map[string][]string {
 	return values
 }

-// extractObjectTags extracts object tags from request (placeholder implementation)
-func extractObjectTags(r *http.Request) []string {
-	// This would need to be implemented based on how object tags are stored
-	// For now, return empty slice
-	return []string{}
-}
-
 // BuildResourceArn builds an ARN for the given bucket and object
 func BuildResourceArn(bucketName, objectName string) string {
 	if objectName == "" {
@@ -352,15 +351,6 @@ func GetObjectNameFromArn(arn string) string {
 	return ""
 }

-// HasPolicyForBucket checks if a bucket has a policy
-func (engine *PolicyEngine) HasPolicyForBucket(bucketName string) bool {
-	engine.mutex.RLock()
-	defer engine.mutex.RUnlock()
-
-	_, exists := engine.contexts[bucketName]
-	return exists
-}
-
 // GetPolicyStatements returns all policy statements for a bucket
 func (engine *PolicyEngine) GetPolicyStatements(bucketName string) []PolicyStatement {
 	engine.mutex.RLock()
@@ -5,9 +5,23 @@ import (
 	"net/url"
 	"testing"

+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 )

+// tagsToEntry converts a map of tag key-value pairs to the entry.Extended format
+// used for s3:ExistingObjectTag/<key> condition evaluation
+func tagsToEntry(tags map[string]string) map[string][]byte {
+	if tags == nil {
+		return nil
+	}
+	entry := make(map[string][]byte)
+	for k, v := range tags {
+		entry[s3_constants.AmzObjectTaggingPrefix+k] = []byte(v)
+	}
+	return entry
+}
+
 func TestPolicyEngine(t *testing.T) {
 	engine := NewPolicyEngine()

@@ -714,3 +728,233 @@ type MockLegacyIAM struct{}
 func (m *MockLegacyIAM) authRequest(r *http.Request, action Action) (Identity, s3err.ErrorCode) {
 	return nil, s3err.ErrNone
 }
+
+// TestExistingObjectTagCondition tests s3:ExistingObjectTag/<tag-key> condition support
+func TestExistingObjectTagCondition(t *testing.T) {
+	engine := NewPolicyEngine()
+
+	// Policy that allows GetObject only for objects with specific tag
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::test-bucket/*",
+				"Condition": {
+					"StringEquals": {
+						"s3:ExistingObjectTag/status": ["public"]
+					}
+				}
+			}
+		]
+	}`
+
+	err := engine.SetBucketPolicy("test-bucket", policyJSON)
+	if err != nil {
+		t.Fatalf("Failed to set bucket policy: %v", err)
+	}
+
+	tests := []struct {
+		name       string
+		objectTags map[string]string
+		expected   PolicyEvaluationResult
+	}{
+		{
+			name:       "Matching tag value - should allow",
+			objectTags: map[string]string{"status": "public"},
+			expected:   PolicyResultAllow,
+		},
+		{
+			name:       "Non-matching tag value - should be indeterminate",
+			objectTags: map[string]string{"status": "private"},
+			expected:   PolicyResultIndeterminate,
+		},
+		{
+			name:       "Missing tag - should be indeterminate",
+			objectTags: map[string]string{"other": "value"},
+			expected:   PolicyResultIndeterminate,
+		},
+		{
+			name:       "No tags - should be indeterminate",
+			objectTags: nil,
+			expected:   PolicyResultIndeterminate,
+		},
+		{
+			name:       "Empty tags - should be indeterminate",
+			objectTags: map[string]string{},
+			expected:   PolicyResultIndeterminate,
+		},
+		{
+			name:       "Multiple tags with matching one - should allow",
+			objectTags: map[string]string{"status": "public", "owner": "admin"},
+			expected:   PolicyResultAllow,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			args := &PolicyEvaluationArgs{
+				Action:      "s3:GetObject",
+				Resource:   "arn:aws:s3:::test-bucket/test-object",
+				Principal:  "*",
+				ObjectEntry: tagsToEntry(tt.objectTags),
+			}
+
+			result := engine.EvaluatePolicy("test-bucket", args)
+			if result != tt.expected {
+				t.Errorf("Expected %v, got %v", tt.expected, result)
+			}
+		})
+	}
+}
+
+// TestExistingObjectTagConditionMultipleTags tests policies with multiple tag conditions
+func TestExistingObjectTagConditionMultipleTags(t *testing.T) {
+	engine := NewPolicyEngine()
+
+	// Policy that requires multiple tag conditions
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::test-bucket/*",
+				"Condition": {
+					"StringEquals": {
+						"s3:ExistingObjectTag/status": ["public"],
+						"s3:ExistingObjectTag/tier": ["free", "premium"]
+					}
+				}
+			}
+		]
+	}`
+
+	err := engine.SetBucketPolicy("test-bucket", policyJSON)
+	if err != nil {
+		t.Fatalf("Failed to set bucket policy: %v", err)
+	}
+
+	tests := []struct {
+		name       string
+		objectTags map[string]string
+		expected   PolicyEvaluationResult
+	}{
+		{
+			name:       "Both tags match - should allow",
+			objectTags: map[string]string{"status": "public", "tier": "free"},
+			expected:   PolicyResultAllow,
+		},
+		{
+			name:       "Both tags match (premium tier) - should allow",
+			objectTags: map[string]string{"status": "public", "tier": "premium"},
+			expected:   PolicyResultAllow,
+		},
+		{
+			name:       "Only status matches - should be indeterminate",
+			objectTags: map[string]string{"status": "public"},
+			expected:   PolicyResultIndeterminate,
+		},
+		{
+			name:       "Only tier matches - should be indeterminate",
+			objectTags: map[string]string{"tier": "free"},
+			expected:   PolicyResultIndeterminate,
+		},
+		{
+			name:       "Neither tag matches - should be indeterminate",
+			objectTags: map[string]string{"status": "private", "tier": "basic"},
+			expected:   PolicyResultIndeterminate,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			args := &PolicyEvaluationArgs{
+				Action:      "s3:GetObject",
+				Resource:   "arn:aws:s3:::test-bucket/test-object",
+				Principal:  "*",
+				ObjectEntry: tagsToEntry(tt.objectTags),
+			}
+
+			result := engine.EvaluatePolicy("test-bucket", args)
+			if result != tt.expected {
+				t.Errorf("Expected %v, got %v", tt.expected, result)
+			}
+		})
+	}
+}
+
+// TestExistingObjectTagDenyPolicy tests deny policies with tag conditions
+func TestExistingObjectTagDenyPolicy(t *testing.T) {
+	engine := NewPolicyEngine()
+
+	// Policy that denies access to objects with confidential tag
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::test-bucket/*"
+			},
+			{
+				"Effect": "Deny",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::test-bucket/*",
+				"Condition": {
+					"StringEquals": {
+						"s3:ExistingObjectTag/classification": ["confidential"]
+					}
+				}
+			}
+		]
+	}`
+
+	err := engine.SetBucketPolicy("test-bucket", policyJSON)
+	if err != nil {
+		t.Fatalf("Failed to set bucket policy: %v", err)
+	}
+
+	tests := []struct {
+		name       string
+		objectTags map[string]string
+		expected   PolicyEvaluationResult
+	}{
+		{
+			name:       "No tags - allow by default statement",
+			objectTags: nil,
+			expected:   PolicyResultAllow,
+		},
+		{
+			name:       "Non-confidential tag - allow",
+			objectTags: map[string]string{"classification": "public"},
+			expected:   PolicyResultAllow,
+		},
+		{
+			name:       "Confidential tag - deny",
+			objectTags: map[string]string{"classification": "confidential"},
+			expected:   PolicyResultDeny,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			args := &PolicyEvaluationArgs{
+				Action:      "s3:GetObject",
+				Resource:   "arn:aws:s3:::test-bucket/test-object",
+				Principal:  "*",
+				ObjectEntry: tagsToEntry(tt.objectTags),
+			}
+
+			result := engine.EvaluatePolicy("test-bucket", args)
+			if result != tt.expected {
+				t.Errorf("Expected %v, got %v", tt.expected, result)
+			}
+		})
+	}
+}
@@ -106,6 +106,11 @@ type PolicyEvaluationArgs struct {
 	Resource   string
 	Principal  string
 	Conditions map[string][]string
+	// ObjectEntry is the object's metadata from entry.Extended.
+	// Used for evaluating conditions like s3:ExistingObjectTag/<tag-key>.
+	// Tags are stored with s3_constants.AmzObjectTaggingPrefix (X-Amz-Tagging-) prefix.
+	// Can be nil for bucket-level operations or when object doesn't exist.
+	ObjectEntry map[string][]byte
 }

 // PolicyCache for caching compiled policies
@@ -0,0 +1,260 @@
+package s3api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestCheckPolicyWithEntry tests the checkPolicyWithEntry helper method
+// This verifies that tag-based conditions are properly evaluated when
+// the object entry is available (Phase 2 of policy evaluation)
+func TestCheckPolicyWithEntry(t *testing.T) {
+	// Create a minimal S3ApiServer with policy engine
+	s3a := &S3ApiServer{
+		policyEngine: NewBucketPolicyEngine(),
+	}
+
+	bucket := "test-bucket"
+
+	// Set up a policy that allows access only to objects with status=public tag
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::test-bucket/*",
+				"Condition": {
+					"StringEquals": {
+						"s3:ExistingObjectTag/status": ["public"]
+					}
+				}
+			}
+		]
+	}`
+
+	err := s3a.policyEngine.engine.SetBucketPolicy(bucket, policyJSON)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name        string
+		objectEntry map[string][]byte
+		wantErrCode s3err.ErrorCode
+		wantChecked bool
+		description string
+	}{
+		{
+			name: "object with public tag - allow",
+			objectEntry: map[string][]byte{
+				s3_constants.AmzObjectTaggingPrefix + "status": []byte("public"),
+			},
+			wantErrCode: s3err.ErrNone,
+			wantChecked: true,
+			description: "Object with status=public tag should be allowed",
+		},
+		{
+			name: "object with private tag - indeterminate (no matching statement)",
+			objectEntry: map[string][]byte{
+				s3_constants.AmzObjectTaggingPrefix + "status": []byte("private"),
+			},
+			wantErrCode: s3err.ErrNone,
+			wantChecked: false, // Indeterminate = no policy matched
+			description: "Object with status=private should return indeterminate (falls back to IAM)",
+		},
+		{
+			name:        "object with no tags - indeterminate (no matching statement)",
+			objectEntry: map[string][]byte{},
+			wantErrCode: s3err.ErrNone,
+			wantChecked: false, // Indeterminate = no policy matched
+			description: "Object without tags should return indeterminate (falls back to IAM)",
+		},
+		{
+			name:        "nil entry - indeterminate (condition cannot be evaluated)",
+			objectEntry: nil,
+			wantErrCode: s3err.ErrNone,
+			wantChecked: false, // Indeterminate = no policy matched
+			description: "Nil entry should return indeterminate (falls back to IAM)",
+		},
+		{
+			name: "object with multiple tags including public - allow",
+			objectEntry: map[string][]byte{
+				s3_constants.AmzObjectTaggingPrefix + "status":   []byte("public"),
+				s3_constants.AmzObjectTaggingPrefix + "category": []byte("documents"),
+			},
+			wantErrCode: s3err.ErrNone,
+			wantChecked: true,
+			description: "Object with status=public among other tags should be allowed",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodGet, "/test-bucket/test-object.txt", nil)
+			object := "/test-object.txt"
+			principal := "*"
+			action := string(s3_constants.ACTION_READ)
+
+			errCode, checked := s3a.checkPolicyWithEntry(req, bucket, object, action, principal, tt.objectEntry)
+
+			assert.Equal(t, tt.wantErrCode, errCode, tt.description)
+			assert.Equal(t, tt.wantChecked, checked, "Policy should be evaluated")
+		})
+	}
+}
+
+// TestCheckPolicyWithEntryNoPolicyForBucket tests that checkPolicyWithEntry
+// returns early when there's no policy for the bucket
+func TestCheckPolicyWithEntryNoPolicyForBucket(t *testing.T) {
+	s3a := &S3ApiServer{
+		policyEngine: NewBucketPolicyEngine(),
+	}
+
+	bucket := "bucket-without-policy"
+	req := httptest.NewRequest(http.MethodGet, "/bucket-without-policy/test.txt", nil)
+
+	objectEntry := map[string][]byte{
+		s3_constants.AmzObjectTaggingPrefix + "status": []byte("private"),
+	}
+
+	errCode, checked := s3a.checkPolicyWithEntry(req, bucket, "/test.txt", string(s3_constants.ACTION_READ), "*", objectEntry)
+
+	assert.Equal(t, s3err.ErrNone, errCode, "No policy should mean no denial")
+	assert.False(t, checked, "Policy should not be evaluated when bucket has no policy")
+}
+
+// TestCheckPolicyWithEntryNilPolicyEngine tests that checkPolicyWithEntry
+// handles nil policy engine gracefully
+func TestCheckPolicyWithEntryNilPolicyEngine(t *testing.T) {
+	s3a := &S3ApiServer{
+		policyEngine: nil,
+	}
+
+	req := httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt", nil)
+
+	errCode, checked := s3a.checkPolicyWithEntry(req, "test-bucket", "/test.txt", string(s3_constants.ACTION_READ), "*", nil)
+
+	assert.Equal(t, s3err.ErrNone, errCode, "Nil policy engine should allow access")
+	assert.False(t, checked, "Policy should not be evaluated when engine is nil")
+}
+
+// TestCheckPolicyWithEntryDenyPolicy tests deny policies with tag conditions
+func TestCheckPolicyWithEntryDenyPolicy(t *testing.T) {
+	s3a := &S3ApiServer{
+		policyEngine: NewBucketPolicyEngine(),
+	}
+
+	bucket := "secure-bucket"
+
+	// Policy that allows all access but denies access to confidential objects
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::secure-bucket/*"
+			},
+			{
+				"Effect": "Deny",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::secure-bucket/*",
+				"Condition": {
+					"StringEquals": {
+						"s3:ExistingObjectTag/classification": ["confidential"]
+					}
+				}
+			}
+		]
+	}`
+
+	err := s3a.policyEngine.engine.SetBucketPolicy(bucket, policyJSON)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name        string
+		objectEntry map[string][]byte
+		wantErrCode s3err.ErrorCode
+		description string
+	}{
+		{
+			name: "public object - allow",
+			objectEntry: map[string][]byte{
+				s3_constants.AmzObjectTaggingPrefix + "classification": []byte("public"),
+			},
+			wantErrCode: s3err.ErrNone,
+			description: "Object with classification=public should be allowed",
+		},
+		{
+			name: "confidential object - deny",
+			objectEntry: map[string][]byte{
+				s3_constants.AmzObjectTaggingPrefix + "classification": []byte("confidential"),
+			},
+			wantErrCode: s3err.ErrAccessDenied,
+			description: "Object with classification=confidential should be denied by explicit deny",
+		},
+		{
+			name:        "no classification tag - allow",
+			objectEntry: map[string][]byte{},
+			wantErrCode: s3err.ErrNone,
+			description: "Object without classification tag should be allowed (deny condition not met)",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodGet, "/secure-bucket/test.txt", nil)
+
+			errCode, checked := s3a.checkPolicyWithEntry(req, bucket, "/test.txt", string(s3_constants.ACTION_READ), "*", tt.objectEntry)
+
+			assert.Equal(t, tt.wantErrCode, errCode, tt.description)
+			assert.True(t, checked, "Policy should be evaluated")
+		})
+	}
+}
+
+// TestHasPolicyForBucket tests the HasPolicyForBucket method
+func TestHasPolicyForBucket(t *testing.T) {
+	engine := policy_engine.NewPolicyEngine()
+
+	// Initially no policy
+	assert.False(t, engine.HasPolicyForBucket("test-bucket"))
+
+	// Set a policy
+	policyJSON := `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Principal": "*",
+				"Action": "s3:GetObject",
+				"Resource": "arn:aws:s3:::test-bucket/*"
+			}
+		]
+	}`
+	err := engine.SetBucketPolicy("test-bucket", policyJSON)
+	require.NoError(t, err)
+
+	// Now has policy
+	assert.True(t, engine.HasPolicyForBucket("test-bucket"))
+
+	// Other bucket still has no policy
+	assert.False(t, engine.HasPolicyForBucket("other-bucket"))
+
+	// Delete the policy
+	err = engine.DeleteBucketPolicy("test-bucket")
+	require.NoError(t, err)
+
+	// No longer has policy
+	assert.False(t, engine.HasPolicyForBucket("test-bucket"))
+}
+
@@ -765,8 +765,8 @@ func (s3a *S3ApiServer) AuthWithPublicRead(handler http.HandlerFunc, action Acti

 			// Check bucket policy for anonymous access using the policy engine
 			principal := "*" // Anonymous principal
-			// Use context-aware policy evaluation to get the correct S3 action
-			allowed, evaluated, err := s3a.policyEngine.EvaluatePolicyWithContext(bucket, object, string(action), principal, r)
+			// Evaluate bucket policy (objectEntry nil - not yet fetched)
+			allowed, evaluated, err := s3a.policyEngine.EvaluatePolicy(bucket, object, string(action), principal, r, nil)
 			if err != nil {
 				// SECURITY: Fail-close on policy evaluation errors
 				// If we can't evaluate the policy, deny access rather than falling through to IAM
@@ -87,12 +87,27 @@ func (bpe *BucketPolicyEngine) DeleteBucketPolicy(bucket string) error {
 	return bpe.engine.DeleteBucketPolicy(bucket)
 }

+// HasPolicyForBucket checks if a bucket has a policy configured
+func (bpe *BucketPolicyEngine) HasPolicyForBucket(bucket string) bool {
+	return bpe.engine.HasPolicyForBucket(bucket)
+}
+
 // EvaluatePolicy evaluates whether an action is allowed by bucket policy
-// Returns: (allowed bool, evaluated bool, error)
-// - allowed: whether the policy allows the action
-// - evaluated: whether a policy was found and evaluated (false = no policy exists)
-// - error: any error during evaluation
-func (bpe *BucketPolicyEngine) EvaluatePolicy(bucket, object, action, principal string) (allowed bool, evaluated bool, err error) {
+//
+// Parameters:
+//   - bucket: the bucket name
+//   - object: the object key (can be empty for bucket-level operations)
+//   - action: the action being performed (e.g., "Read", "Write")
+//   - principal: the principal ARN or identifier
+//   - r: the HTTP request (optional, used for condition evaluation and action resolution)
+//   - objectEntry: the object's metadata from entry.Extended (can be nil at auth time,
+//     should be passed when available for tag-based conditions like s3:ExistingObjectTag)
+//
+// Returns:
+//   - allowed: whether the policy allows the action
+//   - evaluated: whether a policy was found and evaluated (false = no policy exists)
+//   - error: any error during evaluation
+func (bpe *BucketPolicyEngine) EvaluatePolicy(bucket, object, action, principal string, r *http.Request, objectEntry map[string][]byte) (allowed bool, evaluated bool, err error) {
 	// Validate required parameters
 	if bucket == "" {
 		return false, false, fmt.Errorf("bucket cannot be empty")
@@ -101,19 +116,22 @@ func (bpe *BucketPolicyEngine) EvaluatePolicy(bucket, object, action, principal
 		return false, false, fmt.Errorf("action cannot be empty")
 	}

-	// Convert action to S3 action format using base mapping (no HTTP context available)
-	s3Action := mapBaseActionToS3Format(action)
+	// Convert action to S3 action format
+	// ResolveS3Action handles nil request internally (falls back to mapBaseActionToS3Format)
+	s3Action := ResolveS3Action(r, action, bucket, object)

 	// Build resource ARN
 	resource := buildResourceARN(bucket, object)

-	glog.V(4).Infof("EvaluatePolicy: bucket=%s, resource=%s, action=%s, principal=%s", bucket, resource, s3Action, principal)
+	glog.V(4).Infof("EvaluatePolicy: bucket=%s, resource=%s, action=%s, principal=%s",
+		bucket, resource, s3Action, principal)

 	// Evaluate using the policy engine
 	args := &policy_engine.PolicyEvaluationArgs{
-		Action:    s3Action,
-		Resource:  resource,
-		Principal: principal,
+		Action:      s3Action,
+		Resource:    resource,
+		Principal:   principal,
+		ObjectEntry: objectEntry,
 	}

 	result := bpe.engine.EvaluatePolicy(bucket, args)
@@ -133,53 +151,3 @@ func (bpe *BucketPolicyEngine) EvaluatePolicy(bucket, object, action, principal
 		return false, false, fmt.Errorf("unknown policy result: %v", result)
 	}
 }
-
-// EvaluatePolicyWithContext evaluates whether an action is allowed by bucket policy using HTTP request context
-// This version uses the HTTP request to determine the actual S3 action more accurately
-func (bpe *BucketPolicyEngine) EvaluatePolicyWithContext(bucket, object, action, principal string, r *http.Request) (allowed bool, evaluated bool, err error) {
-	// Validate required parameters
-	if bucket == "" {
-		return false, false, fmt.Errorf("bucket cannot be empty")
-	}
-	if action == "" {
-		return false, false, fmt.Errorf("action cannot be empty")
-	}
-
-	// Convert action to S3 action format using request context
-	// ResolveS3Action handles nil request internally (falls back to mapBaseActionToS3Format)
-	s3Action := ResolveS3Action(r, action, bucket, object)
-
-	// Build resource ARN
-	resource := buildResourceARN(bucket, object)
-
-	glog.V(4).Infof("EvaluatePolicyWithContext: bucket=%s, resource=%s, action=%s (from %s), principal=%s",
-		bucket, resource, s3Action, action, principal)
-
-	// Evaluate using the policy engine
-	args := &policy_engine.PolicyEvaluationArgs{
-		Action:    s3Action,
-		Resource:  resource,
-		Principal: principal,
-	}
-
-	result := bpe.engine.EvaluatePolicy(bucket, args)
-
-	switch result {
-	case policy_engine.PolicyResultAllow:
-		glog.V(3).Infof("EvaluatePolicyWithContext: ALLOW - bucket=%s, action=%s, principal=%s", bucket, s3Action, principal)
-		return true, true, nil
-	case policy_engine.PolicyResultDeny:
-		glog.V(3).Infof("EvaluatePolicyWithContext: DENY - bucket=%s, action=%s, principal=%s", bucket, s3Action, principal)
-		return false, true, nil
-	case policy_engine.PolicyResultIndeterminate:
-		// No policy exists for this bucket
-		glog.V(4).Infof("EvaluatePolicyWithContext: INDETERMINATE (no policy) - bucket=%s", bucket)
-		return false, false, nil
-	default:
-		return false, false, fmt.Errorf("unknown policy result: %v", result)
-	}
-}
-
-// NOTE: The convertActionToS3Format wrapper has been removed for simplicity.
-// EvaluatePolicy and EvaluatePolicyWithContext now call ResolveS3Action or
-// mapBaseActionToS3Format directly, making the control flow more explicit.
@@ -634,6 +634,19 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 	}
 	entryFetchTime = time.Since(tEntryFetch)

+	// Safety check: entry must be valid before tag-based policy evaluation
+	if objectEntryForSSE == nil {
+		glog.Errorf("GetObjectHandler: objectEntryForSSE is nil for %s/%s (should not happen)", bucket, object)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		return
+	}
+
+	// Re-check bucket policy with object entry for tag-based conditions (e.g., s3:ExistingObjectTag)
+	if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, string(s3_constants.ACTION_READ), objectEntryForSSE.Extended, "GetObjectHandler"); errCode != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, errCode)
+		return
+	}
+
 	// Check if PartNumber query parameter is present (for multipart GET requests)
 	partNumberStr := r.URL.Query().Get("partNumber")
 	if partNumberStr == "" {
@@ -2187,6 +2200,12 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 		return
 	}

+	// Re-check bucket policy with object entry for tag-based conditions (e.g., s3:ExistingObjectTag)
+	if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, string(s3_constants.ACTION_READ), objectEntryForSSE.Extended, "HeadObjectHandler"); errCode != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, errCode)
+		return
+	}
+
 	// Implicit Directory Handling for s3fs Compatibility
 	// ====================================================
 	//
@@ -573,6 +573,10 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 			break
 		}

+		// Update nextMarker to ensure pagination advances past this .versions directory
+		// This is critical to prevent infinite loops when results are truncated
+		nextMarker = versionsDir
+
 		// Extract object name from .versions directory name (remove .versions suffix)
 		baseObjectName := strings.TrimSuffix(versionsDir, s3_constants.VersionsFolder)

@@ -264,225 +264,237 @@ func (s3a *S3ApiServer) listObjectVersions(bucket, prefix, keyMarker, versionIdM

 // findVersionsRecursively searches for all .versions directories and regular files recursively
 func (s3a *S3ApiServer) findVersionsRecursively(currentPath, relativePath string, allVersions *[]interface{}, processedObjects map[string]bool, seenVersionIds map[string]bool, bucket, prefix string) error {
-	// List entries in current directory
-	entries, _, err := s3a.list(currentPath, "", "", false, 1000)
-	if err != nil {
-		return err
-	}
-
-	for _, entry := range entries {
-		entryPath := path.Join(relativePath, entry.Name)
-
-		// Skip if this doesn't match the prefix filter
-		if normalizedPrefix := strings.TrimPrefix(prefix, "/"); normalizedPrefix != "" {
-			// An entry is a candidate if:
-			// 1. Its path is a match for the prefix.
-			// 2. It is a directory that is an ancestor of the prefix path, so we must descend into it.
-
-			// Condition 1: The entry's path starts with the prefix.
-			isMatch := strings.HasPrefix(entryPath, normalizedPrefix)
-			if !isMatch && entry.IsDirectory {
-				// Also check if a directory entry matches a directory-style prefix (e.g., prefix "a/", entry "a").
-				isMatch = strings.HasPrefix(entryPath+"/", normalizedPrefix)
-			}
-
-			// Condition 2: The prefix path starts with the entry's path (and it's a directory).
-			canDescend := entry.IsDirectory && strings.HasPrefix(normalizedPrefix, entryPath)
-
-			if !isMatch && !canDescend {
-				continue
-			}
+	// List entries in current directory with pagination
+	startFrom := ""
+	for {
+		entries, isLast, err := s3a.list(currentPath, "", startFrom, false, filer.PaginationSize)
+		if err != nil {
+			return err
 		}

-		if entry.IsDirectory {
-			// Skip .uploads directory (multipart upload temporary files)
-			if strings.HasPrefix(entry.Name, ".uploads") {
-				continue
+		for _, entry := range entries {
+			// Track last entry name for pagination
+			startFrom = entry.Name
+
+			entryPath := path.Join(relativePath, entry.Name)
+
+			// Skip if this doesn't match the prefix filter
+			if normalizedPrefix := strings.TrimPrefix(prefix, "/"); normalizedPrefix != "" {
+				// An entry is a candidate if:
+				// 1. Its path is a match for the prefix.
+				// 2. It is a directory that is an ancestor of the prefix path, so we must descend into it.
+
+				// Condition 1: The entry's path starts with the prefix.
+				isMatch := strings.HasPrefix(entryPath, normalizedPrefix)
+				if !isMatch && entry.IsDirectory {
+					// Also check if a directory entry matches a directory-style prefix (e.g., prefix "a/", entry "a").
+					isMatch = strings.HasPrefix(entryPath+"/", normalizedPrefix)
+				}
+
+				// Condition 2: The prefix path starts with the entry's path (and it's a directory).
+				canDescend := entry.IsDirectory && strings.HasPrefix(normalizedPrefix, entryPath)
+
+				if !isMatch && !canDescend {
+					continue
+				}
 			}

-			// Check if this is a .versions directory
-			if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
-				// Extract object name from .versions directory name
-				objectKey := strings.TrimSuffix(entryPath, s3_constants.VersionsFolder)
-				normalizedObjectKey := removeDuplicateSlashes(objectKey)
-				// Mark both keys as processed for backward compatibility
-				processedObjects[objectKey] = true
-				processedObjects[normalizedObjectKey] = true
-
-				glog.V(2).Infof("Found .versions directory for object %s (normalized: %s)", objectKey, normalizedObjectKey)
-
-				versions, err := s3a.getObjectVersionList(bucket, normalizedObjectKey)
-				if err != nil {
-					glog.Warningf("Failed to get versions for object %s (normalized: %s): %v", objectKey, normalizedObjectKey, err)
+			if entry.IsDirectory {
+				// Skip .uploads directory (multipart upload temporary files)
+				if strings.HasPrefix(entry.Name, ".uploads") {
 					continue
 				}

-				for _, version := range versions {
-					// Check for duplicate version IDs and skip if already seen
-					// Use normalized key for deduplication
-					versionKey := normalizedObjectKey + ":" + version.VersionId
-					if seenVersionIds[versionKey] {
-						glog.Warningf("findVersionsRecursively: duplicate version %s for object %s detected, skipping", version.VersionId, normalizedObjectKey)
+				// Check if this is a .versions directory
+				if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
+					// Extract object name from .versions directory name
+					objectKey := strings.TrimSuffix(entryPath, s3_constants.VersionsFolder)
+					normalizedObjectKey := removeDuplicateSlashes(objectKey)
+					// Mark both keys as processed for backward compatibility
+					processedObjects[objectKey] = true
+					processedObjects[normalizedObjectKey] = true
+
+					glog.V(2).Infof("Found .versions directory for object %s (normalized: %s)", objectKey, normalizedObjectKey)
+
+					versions, err := s3a.getObjectVersionList(bucket, normalizedObjectKey)
+					if err != nil {
+						glog.Warningf("Failed to get versions for object %s (normalized: %s): %v", objectKey, normalizedObjectKey, err)
 						continue
 					}
-					seenVersionIds[versionKey] = true

-					if version.IsDeleteMarker {
-						glog.V(4).Infof("Adding delete marker from .versions: objectKey=%s, versionId=%s, isLatest=%v, versionKey=%s",
-							normalizedObjectKey, version.VersionId, version.IsLatest, versionKey)
-						deleteMarker := &DeleteMarkerEntry{
-							Key:          normalizedObjectKey, // Use normalized key for consistency
-							VersionId:    version.VersionId,
-							IsLatest:     version.IsLatest,
-							LastModified: version.LastModified,
-							Owner:        s3a.getObjectOwnerFromVersion(version, bucket, normalizedObjectKey),
+					for _, version := range versions {
+						// Check for duplicate version IDs and skip if already seen
+						// Use normalized key for deduplication
+						versionKey := normalizedObjectKey + ":" + version.VersionId
+						if seenVersionIds[versionKey] {
+							glog.Warningf("findVersionsRecursively: duplicate version %s for object %s detected, skipping", version.VersionId, normalizedObjectKey)
+							continue
 						}
-						*allVersions = append(*allVersions, deleteMarker)
-					} else {
-						glog.V(4).Infof("Adding version from .versions: objectKey=%s, versionId=%s, isLatest=%v, versionKey=%s",
-							normalizedObjectKey, version.VersionId, version.IsLatest, versionKey)
+						seenVersionIds[versionKey] = true
+
+						if version.IsDeleteMarker {
+							glog.V(4).Infof("Adding delete marker from .versions: objectKey=%s, versionId=%s, isLatest=%v, versionKey=%s",
+								normalizedObjectKey, version.VersionId, version.IsLatest, versionKey)
+							deleteMarker := &DeleteMarkerEntry{
+								Key:          normalizedObjectKey, // Use normalized key for consistency
+								VersionId:    version.VersionId,
+								IsLatest:     version.IsLatest,
+								LastModified: version.LastModified,
+								Owner:        s3a.getObjectOwnerFromVersion(version, bucket, normalizedObjectKey),
+							}
+							*allVersions = append(*allVersions, deleteMarker)
+						} else {
+							glog.V(4).Infof("Adding version from .versions: objectKey=%s, versionId=%s, isLatest=%v, versionKey=%s",
+								normalizedObjectKey, version.VersionId, version.IsLatest, versionKey)
+							versionEntry := &VersionEntry{
+								Key:          normalizedObjectKey, // Use normalized key for consistency
+								VersionId:    version.VersionId,
+								IsLatest:     version.IsLatest,
+								LastModified: version.LastModified,
+								ETag:         version.ETag,
+								Size:         version.Size,
+								Owner:        s3a.getObjectOwnerFromVersion(version, bucket, normalizedObjectKey),
+								StorageClass: "STANDARD",
+							}
+							*allVersions = append(*allVersions, versionEntry)
+						}
+					}
+				} else {
+					// This is a regular directory - check if it's an explicit S3 directory object
+					// Only include directories that were explicitly created via S3 API (have FolderMimeType)
+					// This excludes implicit directories created when uploading files like "test1/a"
+					if entry.Attributes.Mime == s3_constants.FolderMimeType {
+						directoryKey := entryPath
+						if !strings.HasSuffix(directoryKey, "/") {
+							directoryKey += "/"
+						}
+
+						// Add directory as a version entry with VersionId "null" (following S3/Minio behavior)
+						glog.V(2).Infof("findVersionsRecursively: found explicit S3 directory %s", directoryKey)
+
+						// Calculate ETag for empty directory
+						directoryETag := "\"d41d8cd98f00b204e9800998ecf8427e\""
+
 						versionEntry := &VersionEntry{
-							Key:          normalizedObjectKey, // Use normalized key for consistency
-							VersionId:    version.VersionId,
-							IsLatest:     version.IsLatest,
-							LastModified: version.LastModified,
-							ETag:         version.ETag,
-							Size:         version.Size,
-							Owner:        s3a.getObjectOwnerFromVersion(version, bucket, normalizedObjectKey),
+							Key:          directoryKey,
+							VersionId:    "null",
+							IsLatest:     true,
+							LastModified: time.Unix(entry.Attributes.Mtime, 0),
+							ETag:         directoryETag,
+							Size:         0, // Directories have size 0
+							Owner:        s3a.getObjectOwnerFromEntry(entry),
 							StorageClass: "STANDARD",
 						}
 						*allVersions = append(*allVersions, versionEntry)
 					}
+
+					// Recursively search subdirectories (regardless of whether they're explicit or implicit)
+					fullPath := path.Join(currentPath, entry.Name)
+					err := s3a.findVersionsRecursively(fullPath, entryPath, allVersions, processedObjects, seenVersionIds, bucket, prefix)
+					if err != nil {
+						glog.Warningf("Error searching subdirectory %s: %v", entryPath, err)
+						continue
+					}
 				}
 			} else {
-				// This is a regular directory - check if it's an explicit S3 directory object
-				// Only include directories that were explicitly created via S3 API (have FolderMimeType)
-				// This excludes implicit directories created when uploading files like "test1/a"
-				if entry.Attributes.Mime == s3_constants.FolderMimeType {
-					directoryKey := entryPath
-					if !strings.HasSuffix(directoryKey, "/") {
-						directoryKey += "/"
-					}
+				// This is a regular file - check if it's a pre-versioning object
+				objectKey := entryPath

-					// Add directory as a version entry with VersionId "null" (following S3/Minio behavior)
-					glog.V(2).Infof("findVersionsRecursively: found explicit S3 directory %s", directoryKey)
+				// Normalize object key to ensure consistency with other version operations
+				normalizedObjectKey := removeDuplicateSlashes(objectKey)

-					// Calculate ETag for empty directory
-					directoryETag := "\"d41d8cd98f00b204e9800998ecf8427e\""
-
-					versionEntry := &VersionEntry{
-						Key:          directoryKey,
-						VersionId:    "null",
-						IsLatest:     true,
-						LastModified: time.Unix(entry.Attributes.Mtime, 0),
-						ETag:         directoryETag,
-						Size:         0, // Directories have size 0
-						Owner:        s3a.getObjectOwnerFromEntry(entry),
-						StorageClass: "STANDARD",
-					}
-					*allVersions = append(*allVersions, versionEntry)
-				}
-
-				// Recursively search subdirectories (regardless of whether they're explicit or implicit)
-				fullPath := path.Join(currentPath, entry.Name)
-				err := s3a.findVersionsRecursively(fullPath, entryPath, allVersions, processedObjects, seenVersionIds, bucket, prefix)
-				if err != nil {
-					glog.Warningf("Error searching subdirectory %s: %v", entryPath, err)
+				// Skip if this object already has a .versions directory (already processed)
+				// Check both normalized and original keys for backward compatibility
+				if processedObjects[objectKey] || processedObjects[normalizedObjectKey] {
+					glog.V(4).Infof("Skipping already processed object: objectKey=%s, normalizedObjectKey=%s, processedObjects[objectKey]=%v, processedObjects[normalizedObjectKey]=%v",
+						objectKey, normalizedObjectKey, processedObjects[objectKey], processedObjects[normalizedObjectKey])
 					continue
 				}
-			}
-		} else {
-			// This is a regular file - check if it's a pre-versioning object
-			objectKey := entryPath

-			// Normalize object key to ensure consistency with other version operations
-			normalizedObjectKey := removeDuplicateSlashes(objectKey)
+				glog.V(4).Infof("Processing regular file: objectKey=%s, normalizedObjectKey=%s, NOT in processedObjects", objectKey, normalizedObjectKey)

-			// Skip if this object already has a .versions directory (already processed)
-			// Check both normalized and original keys for backward compatibility
-			if processedObjects[objectKey] || processedObjects[normalizedObjectKey] {
-				glog.V(4).Infof("Skipping already processed object: objectKey=%s, normalizedObjectKey=%s, processedObjects[objectKey]=%v, processedObjects[normalizedObjectKey]=%v",
-					objectKey, normalizedObjectKey, processedObjects[objectKey], processedObjects[normalizedObjectKey])
-				continue
-			}
-
-			glog.V(4).Infof("Processing regular file: objectKey=%s, normalizedObjectKey=%s, NOT in processedObjects", objectKey, normalizedObjectKey)
-
-			// This is a pre-versioning or suspended-versioning object
-			// Check if this file has version metadata (ExtVersionIdKey)
-			hasVersionMeta := false
-			if entry.Extended != nil {
-				if versionIdBytes, ok := entry.Extended[s3_constants.ExtVersionIdKey]; ok {
-					hasVersionMeta = true
-					glog.V(4).Infof("Regular file %s has version metadata: %s", normalizedObjectKey, string(versionIdBytes))
+				// This is a pre-versioning or suspended-versioning object
+				// Check if this file has version metadata (ExtVersionIdKey)
+				hasVersionMeta := false
+				if entry.Extended != nil {
+					if versionIdBytes, ok := entry.Extended[s3_constants.ExtVersionIdKey]; ok {
+						hasVersionMeta = true
+						glog.V(4).Infof("Regular file %s has version metadata: %s", normalizedObjectKey, string(versionIdBytes))
+					}
 				}
-			}

-			// Check if a .versions directory exists for this object
-			versionsObjectPath := normalizedObjectKey + s3_constants.VersionsFolder
-			_, versionsErr := s3a.getEntry(currentPath, versionsObjectPath)
-			if versionsErr == nil {
-				// .versions directory exists
-				glog.V(4).Infof("Found .versions directory for regular file %s, hasVersionMeta=%v", normalizedObjectKey, hasVersionMeta)
+				// Check if a .versions directory exists for this object
+				// Use entry.Name (relative to currentPath) to avoid duplicating subdirectory segments
+				versionsEntryName := entry.Name + s3_constants.VersionsFolder
+				_, versionsErr := s3a.getEntry(currentPath, versionsEntryName)
+				if versionsErr == nil {
+					// .versions directory exists
+					glog.V(4).Infof("Found .versions directory for regular file %s, hasVersionMeta=%v", normalizedObjectKey, hasVersionMeta)

-				// If this file has version metadata, it's a suspended versioning null version
-				// Include it and it will be the latest
-				if hasVersionMeta {
-					glog.V(4).Infof("Including suspended versioning file %s (has version metadata)", normalizedObjectKey)
-					// Continue to add it below
-				} else {
-					// No version metadata - this is a pre-versioning file
-					// Skip it if there's already a null version in .versions
-					versions, err := s3a.getObjectVersionList(bucket, normalizedObjectKey)
-					if err == nil {
-						hasNullVersion := false
-						for _, v := range versions {
-							if v.VersionId == "null" {
-								hasNullVersion = true
-								break
+					// If this file has version metadata, it's a suspended versioning null version
+					// Include it and it will be the latest
+					if hasVersionMeta {
+						glog.V(4).Infof("Including suspended versioning file %s (has version metadata)", normalizedObjectKey)
+						// Continue to add it below
+					} else {
+						// No version metadata - this is a pre-versioning file
+						// Skip it if there's already a null version in .versions
+						versions, err := s3a.getObjectVersionList(bucket, normalizedObjectKey)
+						if err == nil {
+							hasNullVersion := false
+							for _, v := range versions {
+								if v.VersionId == "null" {
+									hasNullVersion = true
+									break
+								}
+							}
+							if hasNullVersion {
+								glog.V(4).Infof("Skipping pre-versioning file %s, null version exists in .versions", normalizedObjectKey)
+								processedObjects[objectKey] = true
+								processedObjects[normalizedObjectKey] = true
+								continue
 							}
 						}
-						if hasNullVersion {
-							glog.V(4).Infof("Skipping pre-versioning file %s, null version exists in .versions", normalizedObjectKey)
-							processedObjects[objectKey] = true
-							processedObjects[normalizedObjectKey] = true
-							continue
-						}
+						glog.V(4).Infof("Including pre-versioning file %s (no null version in .versions)", normalizedObjectKey)
 					}
-					glog.V(4).Infof("Including pre-versioning file %s (no null version in .versions)", normalizedObjectKey)
+				} else {
+					glog.V(4).Infof("No .versions directory for regular file %s, hasVersionMeta=%v", normalizedObjectKey, hasVersionMeta)
 				}
-			} else {
-				glog.V(4).Infof("No .versions directory for regular file %s, hasVersionMeta=%v", normalizedObjectKey, hasVersionMeta)
+
+				// Add this file as a null version with IsLatest=true
+				isLatest := true
+
+				// Check for duplicate version IDs and skip if already seen
+				// Use normalized key for deduplication to match how other version operations work
+				versionKey := normalizedObjectKey + ":null"
+				if seenVersionIds[versionKey] {
+					glog.Warningf("findVersionsRecursively: duplicate null version for object %s detected (versionKey=%s), skipping", normalizedObjectKey, versionKey)
+					continue
+				}
+				seenVersionIds[versionKey] = true
+
+				etag := s3a.calculateETagFromChunks(entry.Chunks)
+
+				glog.V(4).Infof("Adding null version from regular file: objectKey=%s, normalizedObjectKey=%s, versionKey=%s, isLatest=%v, hasVersionMeta=%v",
+					objectKey, normalizedObjectKey, versionKey, isLatest, hasVersionMeta)
+
+				versionEntry := &VersionEntry{
+					Key:          normalizedObjectKey, // Use normalized key for consistency
+					VersionId:    "null",
+					IsLatest:     isLatest,
+					LastModified: time.Unix(entry.Attributes.Mtime, 0),
+					ETag:         etag,
+					Size:         int64(entry.Attributes.FileSize),
+					Owner:        s3a.getObjectOwnerFromEntry(entry),
+					StorageClass: "STANDARD",
+				}
+				*allVersions = append(*allVersions, versionEntry)
 			}
+		}

-			// Add this file as a null version with IsLatest=true
-			isLatest := true
-
-			// Check for duplicate version IDs and skip if already seen
-			// Use normalized key for deduplication to match how other version operations work
-			versionKey := normalizedObjectKey + ":null"
-			if seenVersionIds[versionKey] {
-				glog.Warningf("findVersionsRecursively: duplicate null version for object %s detected (versionKey=%s), skipping", normalizedObjectKey, versionKey)
-				continue
-			}
-			seenVersionIds[versionKey] = true
-
-			etag := s3a.calculateETagFromChunks(entry.Chunks)
-
-			glog.V(4).Infof("Adding null version from regular file: objectKey=%s, normalizedObjectKey=%s, versionKey=%s, isLatest=%v, hasVersionMeta=%v",
-				objectKey, normalizedObjectKey, versionKey, isLatest, hasVersionMeta)
-
-			versionEntry := &VersionEntry{
-				Key:          normalizedObjectKey, // Use normalized key for consistency
-				VersionId:    "null",
-				IsLatest:     isLatest,
-				LastModified: time.Unix(entry.Attributes.Mtime, 0),
-				ETag:         etag,
-				Size:         int64(entry.Attributes.FileSize),
-				Owner:        s3a.getObjectOwnerFromEntry(entry),
-				StorageClass: "STANDARD",
-			}
-			*allVersions = append(*allVersions, versionEntry)
+		// If we've reached the last page, stop pagination
+		if isLast {
+			break
 		}
 	}

@@ -17,13 +17,8 @@ func getRequestDataReader(s3a *S3ApiServer, r *http.Request) (io.ReadCloser, s3e
 	dataReader := r.Body
 	rAuthType := getRequestAuthType(r)
 	if s3a.iam.isEnabled() {
-		switch rAuthType {
-		case authTypeStreamingSigned, authTypeStreamingUnsigned:
+		if rAuthType == authTypeStreamingSigned || rAuthType == authTypeStreamingUnsigned {
 			dataReader, s3ErrCode = s3a.iam.newChunkedReader(r)
-		case authTypeSignedV2, authTypePresignedV2:
-			_, s3ErrCode = s3a.iam.isReqAuthenticatedV2(r)
-		case authTypePresigned, authTypeSigned:
-			_, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
 		}
 	} else {
 		switch rAuthType {
@@ -252,6 +252,62 @@ func (s3a *S3ApiServer) syncBucketPolicyToEngine(bucket string, policyDoc *polic
 	}
 }

+// checkPolicyWithEntry re-evaluates bucket policy with the object entry metadata.
+// This is used by handlers after fetching the entry to enforce tag-based conditions
+// like s3:ExistingObjectTag/<key>.
+//
+// Returns:
+//   - s3err.ErrCode: ErrNone if allowed, ErrAccessDenied if denied
+//   - bool: true if policy was evaluated (has policy for bucket), false if no policy
+func (s3a *S3ApiServer) checkPolicyWithEntry(r *http.Request, bucket, object, action, principal string, objectEntry map[string][]byte) (s3err.ErrorCode, bool) {
+	if s3a.policyEngine == nil {
+		return s3err.ErrNone, false
+	}
+
+	// Skip if no policy for this bucket
+	if !s3a.policyEngine.HasPolicyForBucket(bucket) {
+		return s3err.ErrNone, false
+	}
+
+	allowed, evaluated, err := s3a.policyEngine.EvaluatePolicy(bucket, object, action, principal, r, objectEntry)
+	if err != nil {
+		glog.Errorf("checkPolicyWithEntry: error evaluating policy for %s/%s: %v", bucket, object, err)
+		return s3err.ErrInternalError, true
+	}
+
+	if !evaluated {
+		return s3err.ErrNone, false
+	}
+
+	if !allowed {
+		glog.V(3).Infof("checkPolicyWithEntry: policy denied access to %s/%s for principal %s", bucket, object, principal)
+		return s3err.ErrAccessDenied, true
+	}
+
+	return s3err.ErrNone, true
+}
+
+// recheckPolicyWithObjectEntry performs the second phase of policy evaluation after
+// an object's entry is fetched. It extracts identity from context and checks for
+// tag-based conditions like s3:ExistingObjectTag/<key>.
+//
+// Returns s3err.ErrNone if allowed, or an error code if denied or on error.
+func (s3a *S3ApiServer) recheckPolicyWithObjectEntry(r *http.Request, bucket, object, action string, objectEntry map[string][]byte, handlerName string) s3err.ErrorCode {
+	identityRaw := GetIdentityFromContext(r)
+	var identity *Identity
+	if identityRaw != nil {
+		var ok bool
+		identity, ok = identityRaw.(*Identity)
+		if !ok {
+			glog.Errorf("%s: unexpected identity type in context for %s/%s", handlerName, bucket, object)
+			return s3err.ErrInternalError
+		}
+	}
+	principal := buildPrincipalARN(identity)
+	errCode, _ := s3a.checkPolicyWithEntry(r, bucket, object, action, principal, objectEntry)
+	return errCode
+}
+
 // classifyDomainNames classifies domains into path-style and virtual-host style domains.
 // A domain is considered path-style if:
 //  1. It contains a dot (has subdomains)
@@ -284,15 +284,30 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 		}
 		ms.deleteClient(clientName)
 	}()
-	for i, message := range ms.Topo.ToVolumeLocations() {
-		if i == 0 {
-			if leader, err := ms.Topo.Leader(); err == nil {
-				message.Leader = string(leader)
-			}
-		}
-		if sendErr := stream.Send(&master_pb.KeepConnectedResponse{VolumeLocation: message}); sendErr != nil {
+
+	// Send volume locations to the client
+	volumeLocations := ms.Topo.ToVolumeLocations()
+	if len(volumeLocations) == 0 {
+		// Always send at least one message with leader info so the client can unblock
+		leader, _ := ms.Topo.Leader()
+		if sendErr := stream.Send(&master_pb.KeepConnectedResponse{
+			VolumeLocation: &master_pb.VolumeLocation{
+				Leader: string(leader),
+			},
+		}); sendErr != nil {
 			return sendErr
 		}
+	} else {
+		for i, message := range volumeLocations {
+			if i == 0 {
+				if leader, err := ms.Topo.Leader(); err == nil {
+					message.Leader = string(leader)
+				}
+			}
+			if sendErr := stream.Send(&master_pb.KeepConnectedResponse{VolumeLocation: message}); sendErr != nil {
+				return sendErr
+			}
+		}
 	}

 	go func() {
@@ -291,6 +291,15 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 		}
 		wt.MaybeSlowdown(int64(len(resp.FileContent)))
 	}
+	// If no data was written (source file was not found), remove the empty file
+	// to avoid leaving corrupted empty files that cause parse errors later
+	if progressedBytes == 0 && !isAppend {
+		if removeErr := os.Remove(fileName); removeErr != nil {
+			glog.V(1).Infof("failed to remove empty file %s: %v", fileName, removeErr)
+		} else {
+			glog.V(1).Infof("removed empty file %s (source file not found)", fileName)
+		}
+	}
 	return modifiedTsNs, nil
 }

@@ -4,6 +4,8 @@ import (
 	"flag"
 	"fmt"
 	"io"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )

 func init() {
@@ -20,7 +22,10 @@ func (c *commandEcBalance) Name() string {
 func (c *commandEcBalance) Help() string {
 	return `balance all ec shards among all racks and volume servers

-	ec.balance [-c EACH_COLLECTION|<collection_name>] [-apply] [-dataCenter <data_center>] [-shardReplicaPlacement <replica_placement>]
+	ec.balance [-c EACH_COLLECTION|<collection_name>] [-apply] [-dataCenter <data_center>] [-shardReplicaPlacement <replica_placement>] [-diskType <disk_type>]
+
+	Options:
+	  -diskType: the disk type for EC shards (hdd, ssd, or empty for default hdd)

 	Algorithm:
 	` + ecBalanceAlgorithmDescription
@@ -35,6 +40,7 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W
 	collection := balanceCommand.String("collection", "EACH_COLLECTION", "collection name, or \"EACH_COLLECTION\" for each collection")
 	dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
 	shardReplicaPlacement := balanceCommand.String("shardReplicaPlacement", "", "replica placement for EC shards, or master default if empty")
+	diskTypeStr := balanceCommand.String("diskType", "", "the disk type for EC shards (hdd, ssd, or empty for default hdd)")
 	maxParallelization := balanceCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
 	applyBalancing := balanceCommand.Bool("apply", false, "apply the balancing plan")
 	// TODO: remove this alias
@@ -67,5 +73,7 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W
 		return err
 	}

-	return EcBalance(commandEnv, collections, *dc, rp, *maxParallelization, *applyBalancing)
+	diskType := types.ToDiskType(*diskTypeStr)
+
+	return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *applyBalancing)
 }
@@ -182,7 +182,7 @@ func collectTopologyInfo(commandEnv *CommandEnv, delayBeforeCollecting time.Dura

 }

-func collectEcNodesForDC(commandEnv *CommandEnv, selectedDataCenter string) (ecNodes []*EcNode, totalFreeEcSlots int, err error) {
+func collectEcNodesForDC(commandEnv *CommandEnv, selectedDataCenter string, diskType types.DiskType) (ecNodes []*EcNode, totalFreeEcSlots int, err error) {
 	// list all possible locations
 	// collect topology information
 	topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
@@ -191,15 +191,15 @@ func collectEcNodesForDC(commandEnv *CommandEnv, selectedDataCenter string) (ecN
 	}

 	// find out all volume servers with one slot left.
-	ecNodes, totalFreeEcSlots = collectEcVolumeServersByDc(topologyInfo, selectedDataCenter)
+	ecNodes, totalFreeEcSlots = collectEcVolumeServersByDc(topologyInfo, selectedDataCenter, diskType)

 	sortEcNodesByFreeslotsDescending(ecNodes)

 	return
 }

-func collectEcNodes(commandEnv *CommandEnv) (ecNodes []*EcNode, totalFreeEcSlots int, err error) {
-	return collectEcNodesForDC(commandEnv, "")
+func collectEcNodes(commandEnv *CommandEnv, diskType types.DiskType) (ecNodes []*EcNode, totalFreeEcSlots int, err error) {
+	return collectEcNodesForDC(commandEnv, "", diskType)
 }

 func collectCollectionsForVolumeIds(t *master_pb.TopologyInfo, vids []needle.VolumeId) []string {
@@ -242,7 +242,7 @@ func collectCollectionsForVolumeIds(t *master_pb.TopologyInfo, vids []needle.Vol
 	return collections
 }

-func moveMountedShardToEcNode(commandEnv *CommandEnv, existingLocation *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, destinationEcNode *EcNode, destDiskId uint32, applyBalancing bool) (err error) {
+func moveMountedShardToEcNode(commandEnv *CommandEnv, existingLocation *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, destinationEcNode *EcNode, destDiskId uint32, applyBalancing bool, diskType types.DiskType) (err error) {

 	if !commandEnv.isLocked() {
 		return fmt.Errorf("lock is lost")
@@ -280,8 +280,8 @@ func moveMountedShardToEcNode(commandEnv *CommandEnv, existingLocation *EcNode,

 	}

-	destinationEcNode.addEcVolumeShards(vid, collection, copiedShardIds)
-	existingLocation.deleteEcVolumeShards(vid, copiedShardIds)
+	destinationEcNode.addEcVolumeShards(vid, collection, copiedShardIds, diskType)
+	existingLocation.deleteEcVolumeShards(vid, copiedShardIds, diskType)

 	return nil

@@ -421,13 +421,13 @@ func (ecNode *EcNode) localShardIdCount(vid uint32) int {
 	return 0
 }

-func collectEcVolumeServersByDc(topo *master_pb.TopologyInfo, selectedDataCenter string) (ecNodes []*EcNode, totalFreeEcSlots int) {
+func collectEcVolumeServersByDc(topo *master_pb.TopologyInfo, selectedDataCenter string, diskType types.DiskType) (ecNodes []*EcNode, totalFreeEcSlots int) {
 	eachDataNode(topo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
 		if selectedDataCenter != "" && selectedDataCenter != string(dc) {
 			return
 		}

-		freeEcSlots := countFreeShardSlots(dn, types.HardDriveType)
+		freeEcSlots := countFreeShardSlots(dn, diskType)
 		ecNode := &EcNode{
 			info:       dn,
 			dc:         dc,
@@ -439,17 +439,17 @@ func collectEcVolumeServersByDc(topo *master_pb.TopologyInfo, selectedDataCenter
 		// Build disk-level information from volumes and EC shards
 		// First, discover all unique disk IDs from VolumeInfos (includes empty disks)
 		allDiskIds := make(map[uint32]string) // diskId -> diskType
-		for diskType, diskInfo := range dn.DiskInfos {
+		for diskTypeKey, diskInfo := range dn.DiskInfos {
 			if diskInfo == nil {
 				continue
 			}
 			// Get all disk IDs from volumes
 			for _, vi := range diskInfo.VolumeInfos {
-				allDiskIds[vi.DiskId] = diskType
+				allDiskIds[vi.DiskId] = diskTypeKey
 			}
 			// Also get disk IDs from EC shards
 			for _, ecShardInfo := range diskInfo.EcShardInfos {
-				allDiskIds[ecShardInfo.DiskId] = diskType
+				allDiskIds[ecShardInfo.DiskId] = diskTypeKey
 			}
 		}

@@ -476,7 +476,7 @@ func collectEcVolumeServersByDc(topo *master_pb.TopologyInfo, selectedDataCenter
 		}
 		freePerDisk := int(freeEcSlots) / diskCount

-		for diskId, diskType := range allDiskIds {
+		for diskId, diskTypeStr := range allDiskIds {
 			shards := diskShards[diskId]
 			if shards == nil {
 				shards = make(map[needle.VolumeId]erasure_coding.ShardBits)
@@ -488,7 +488,7 @@ func collectEcVolumeServersByDc(topo *master_pb.TopologyInfo, selectedDataCenter

 			ecNode.disks[diskId] = &EcDisk{
 				diskId:       diskId,
-				diskType:     diskType,
+				diskType:     diskTypeStr,
 				freeEcSlots:  freePerDisk,
 				ecShardCount: totalShardCount,
 				ecShards:     shards,
@@ -551,9 +551,9 @@ func ceilDivide(a, b int) int {
 	return (a / b) + r
 }

-func findEcVolumeShards(ecNode *EcNode, vid needle.VolumeId) erasure_coding.ShardBits {
+func findEcVolumeShards(ecNode *EcNode, vid needle.VolumeId, diskType types.DiskType) erasure_coding.ShardBits {

-	if diskInfo, found := ecNode.info.DiskInfos[string(types.HardDriveType)]; found {
+	if diskInfo, found := ecNode.info.DiskInfos[string(diskType)]; found {
 		for _, shardInfo := range diskInfo.EcShardInfos {
 			if needle.VolumeId(shardInfo.Id) == vid {
 				return erasure_coding.ShardBits(shardInfo.EcIndexBits)
@@ -564,10 +564,10 @@ func findEcVolumeShards(ecNode *EcNode, vid needle.VolumeId) erasure_coding.Shar
 	return 0
 }

-func (ecNode *EcNode) addEcVolumeShards(vid needle.VolumeId, collection string, shardIds []uint32) *EcNode {
+func (ecNode *EcNode) addEcVolumeShards(vid needle.VolumeId, collection string, shardIds []uint32, diskType types.DiskType) *EcNode {

 	foundVolume := false
-	diskInfo, found := ecNode.info.DiskInfos[string(types.HardDriveType)]
+	diskInfo, found := ecNode.info.DiskInfos[string(diskType)]
 	if found {
 		for _, shardInfo := range diskInfo.EcShardInfos {
 			if needle.VolumeId(shardInfo.Id) == vid {
@@ -584,9 +584,9 @@ func (ecNode *EcNode) addEcVolumeShards(vid needle.VolumeId, collection string,
 		}
 	} else {
 		diskInfo = &master_pb.DiskInfo{
-			Type: string(types.HardDriveType),
+			Type: string(diskType),
 		}
-		ecNode.info.DiskInfos[string(types.HardDriveType)] = diskInfo
+		ecNode.info.DiskInfos[string(diskType)] = diskInfo
 	}

 	if !foundVolume {
@@ -598,7 +598,7 @@ func (ecNode *EcNode) addEcVolumeShards(vid needle.VolumeId, collection string,
 			Id:          uint32(vid),
 			Collection:  collection,
 			EcIndexBits: uint32(newShardBits),
-			DiskType:    string(types.HardDriveType),
+			DiskType:    string(diskType),
 		})
 		ecNode.freeEcSlot -= len(shardIds)
 	}
@@ -606,9 +606,9 @@ func (ecNode *EcNode) addEcVolumeShards(vid needle.VolumeId, collection string,
 	return ecNode
 }

-func (ecNode *EcNode) deleteEcVolumeShards(vid needle.VolumeId, shardIds []uint32) *EcNode {
+func (ecNode *EcNode) deleteEcVolumeShards(vid needle.VolumeId, shardIds []uint32, diskType types.DiskType) *EcNode {

-	if diskInfo, found := ecNode.info.DiskInfos[string(types.HardDriveType)]; found {
+	if diskInfo, found := ecNode.info.DiskInfos[string(diskType)]; found {
 		for _, shardInfo := range diskInfo.EcShardInfos {
 			if needle.VolumeId(shardInfo.Id) == vid {
 				oldShardBits := erasure_coding.ShardBits(shardInfo.EcIndexBits)
@@ -649,6 +649,7 @@ type ecBalancer struct {
 	replicaPlacement   *super_block.ReplicaPlacement
 	applyBalancing     bool
 	maxParallelization int
+	diskType           types.DiskType // target disk type for EC shards (default: HardDriveType)
 }

 func (ecb *ecBalancer) errorWaitGroup() *ErrorWaitGroup {
@@ -705,7 +706,7 @@ func (ecb *ecBalancer) doDeduplicateEcShards(collection string, vid needle.Volum
 	// Use MaxShardCount (32) to support custom EC ratios
 	shardToLocations := make([][]*EcNode, erasure_coding.MaxShardCount)
 	for _, ecNode := range locations {
-		shardBits := findEcVolumeShards(ecNode, vid)
+		shardBits := findEcVolumeShards(ecNode, vid, ecb.diskType)
 		for _, shardId := range shardBits.ShardIds() {
 			shardToLocations[shardId] = append(shardToLocations[shardId], ecNode)
 		}
@@ -728,7 +729,7 @@ func (ecb *ecBalancer) doDeduplicateEcShards(collection string, vid needle.Volum
 			if err := sourceServerDeleteEcShards(ecb.commandEnv.option.GrpcDialOption, collection, vid, pb.NewServerAddressFromDataNode(ecNode.info), duplicatedShardIds); err != nil {
 				return err
 			}
-			ecNode.deleteEcVolumeShards(vid, duplicatedShardIds)
+			ecNode.deleteEcVolumeShards(vid, duplicatedShardIds, ecb.diskType)
 		}
 	}
 	return nil
@@ -748,9 +749,9 @@ func (ecb *ecBalancer) balanceEcShardsAcrossRacks(collection string) error {
 	return ewg.Wait()
 }

-func countShardsByRack(vid needle.VolumeId, locations []*EcNode) map[string]int {
+func countShardsByRack(vid needle.VolumeId, locations []*EcNode, diskType types.DiskType) map[string]int {
 	return groupByCount(locations, func(ecNode *EcNode) (id string, count int) {
-		shardBits := findEcVolumeShards(ecNode, vid)
+		shardBits := findEcVolumeShards(ecNode, vid, diskType)
 		return string(ecNode.rack), shardBits.ShardIdCount()
 	})
 }
@@ -759,7 +760,7 @@ func (ecb *ecBalancer) doBalanceEcShardsAcrossRacks(collection string, vid needl
 	racks := ecb.racks()

 	// see the volume's shards are in how many racks, and how many in each rack
-	rackToShardCount := countShardsByRack(vid, locations)
+	rackToShardCount := countShardsByRack(vid, locations, ecb.diskType)

 	// Calculate actual total shards for this volume (not hardcoded default)
 	var totalShardsForVolume int
@@ -779,7 +780,7 @@ func (ecb *ecBalancer) doBalanceEcShardsAcrossRacks(collection string, vid needl
 			continue
 		}
 		possibleEcNodes := rackEcNodesWithVid[rackId]
-		for shardId, ecNode := range pickNEcShardsToMoveFrom(possibleEcNodes, vid, count-averageShardsPerEcRack) {
+		for shardId, ecNode := range pickNEcShardsToMoveFrom(possibleEcNodes, vid, count-averageShardsPerEcRack, ecb.diskType) {
 			ecShardsToMove[shardId] = ecNode
 		}
 	}
@@ -856,7 +857,7 @@ func (ecb *ecBalancer) balanceEcShardsWithinRacks(collection string) error {
 	for vid, locations := range vidLocations {

 		// see the volume's shards are in how many racks, and how many in each rack
-		rackToShardCount := countShardsByRack(vid, locations)
+		rackToShardCount := countShardsByRack(vid, locations, ecb.diskType)
 		rackEcNodesWithVid := groupBy(locations, func(ecNode *EcNode) string {
 			return string(ecNode.rack)
 		})
@@ -865,7 +866,7 @@ func (ecb *ecBalancer) balanceEcShardsWithinRacks(collection string) error {

 			var possibleDestinationEcNodes []*EcNode
 			for _, n := range racks[RackId(rackId)].ecNodes {
-				if _, found := n.info.DiskInfos[string(types.HardDriveType)]; found {
+				if _, found := n.info.DiskInfos[string(ecb.diskType)]; found {
 					possibleDestinationEcNodes = append(possibleDestinationEcNodes, n)
 				}
 			}
@@ -882,7 +883,7 @@ func (ecb *ecBalancer) balanceEcShardsWithinRacks(collection string) error {
 func (ecb *ecBalancer) doBalanceEcShardsWithinOneRack(averageShardsPerEcNode int, collection string, vid needle.VolumeId, existingLocations, possibleDestinationEcNodes []*EcNode) error {
 	for _, ecNode := range existingLocations {

-		shardBits := findEcVolumeShards(ecNode, vid)
+		shardBits := findEcVolumeShards(ecNode, vid, ecb.diskType)
 		overLimitCount := shardBits.ShardIdCount() - averageShardsPerEcNode

 		for _, shardId := range shardBits.ShardIds() {
@@ -927,7 +928,7 @@ func (ecb *ecBalancer) doBalanceEcRack(ecRack *EcRack) error {
 	}

 	ecNodeIdToShardCount := groupByCount(rackEcNodes, func(ecNode *EcNode) (id string, count int) {
-		diskInfo, found := ecNode.info.DiskInfos[string(types.HardDriveType)]
+		diskInfo, found := ecNode.info.DiskInfos[string(ecb.diskType)]
 		if !found {
 			return
 		}
@@ -955,17 +956,18 @@ func (ecb *ecBalancer) doBalanceEcRack(ecRack *EcRack) error {
 		if fullNodeShardCount > averageShardCount && emptyNodeShardCount+1 <= averageShardCount {

 			emptyNodeIds := make(map[uint32]bool)
-			if emptyDiskInfo, found := emptyNode.info.DiskInfos[string(types.HardDriveType)]; found {
+			if emptyDiskInfo, found := emptyNode.info.DiskInfos[string(ecb.diskType)]; found {
 				for _, shards := range emptyDiskInfo.EcShardInfos {
 					emptyNodeIds[shards.Id] = true
 				}
 			}
-			if fullDiskInfo, found := fullNode.info.DiskInfos[string(types.HardDriveType)]; found {
+			if fullDiskInfo, found := fullNode.info.DiskInfos[string(ecb.diskType)]; found {
 				for _, shards := range fullDiskInfo.EcShardInfos {
 					if _, found := emptyNodeIds[shards.Id]; !found {
 						for _, shardId := range erasure_coding.ShardBits(shards.EcIndexBits).ShardIds() {
 							vid := needle.VolumeId(shards.Id)
-							destDiskId := pickBestDiskOnNode(emptyNode, vid)
+							// For balancing, strictly require matching disk type
+							destDiskId := pickBestDiskOnNode(emptyNode, vid, ecb.diskType, true)

 							if destDiskId > 0 {
 								fmt.Printf("%s moves ec shards %d.%d to %s (disk %d)\n", fullNode.info.Id, shards.Id, shardId, emptyNode.info.Id, destDiskId)
@@ -973,7 +975,7 @@ func (ecb *ecBalancer) doBalanceEcRack(ecRack *EcRack) error {
 								fmt.Printf("%s moves ec shards %d.%d to %s\n", fullNode.info.Id, shards.Id, shardId, emptyNode.info.Id)
 							}

-							err := moveMountedShardToEcNode(ecb.commandEnv, fullNode, shards.Collection, vid, shardId, emptyNode, destDiskId, ecb.applyBalancing)
+							err := moveMountedShardToEcNode(ecb.commandEnv, fullNode, shards.Collection, vid, shardId, emptyNode, destDiskId, ecb.applyBalancing, ecb.diskType)
 							if err != nil {
 								return err
 							}
@@ -1003,7 +1005,7 @@ func (ecb *ecBalancer) pickEcNodeToBalanceShardsInto(vid needle.VolumeId, existi

 	nodeShards := map[*EcNode]int{}
 	for _, node := range possibleDestinations {
-		nodeShards[node] = findEcVolumeShards(node, vid).ShardIdCount()
+		nodeShards[node] = findEcVolumeShards(node, vid, ecb.diskType).ShardIdCount()
 	}

 	targets := []*EcNode{}
@@ -1078,14 +1080,17 @@ func diskDistributionScore(ecNode *EcNode, vid needle.VolumeId) int {
 }

 // pickBestDiskOnNode selects the best disk on a node for placing a new EC shard
-// It prefers disks with fewer shards and more free slots
-func pickBestDiskOnNode(ecNode *EcNode, vid needle.VolumeId) uint32 {
+// It prefers disks of the specified type with fewer shards and more free slots
+// If strictDiskType is false, it will fall back to other disk types if no matching disk is found
+func pickBestDiskOnNode(ecNode *EcNode, vid needle.VolumeId, diskType types.DiskType, strictDiskType bool) uint32 {
 	if len(ecNode.disks) == 0 {
 		return 0 // No disk info available, let the server decide
 	}

 	var bestDiskId uint32
 	bestScore := -1
+	var fallbackDiskId uint32
+	fallbackScore := -1

 	for diskId, disk := range ecNode.disks {
 		if disk.freeEcSlots <= 0 {
@@ -1102,13 +1107,26 @@ func pickBestDiskOnNode(ecNode *EcNode, vid needle.VolumeId) uint32 {
 		// Lower score is better
 		score := disk.ecShardCount*10 + existingShards*100

-		if bestScore == -1 || score < bestScore {
-			bestScore = score
-			bestDiskId = diskId
+		if disk.diskType == string(diskType) {
+			// Matching disk type - this is preferred
+			if bestScore == -1 || score < bestScore {
+				bestScore = score
+				bestDiskId = diskId
+			}
+		} else if !strictDiskType {
+			// Non-matching disk type - use as fallback if allowed
+			if fallbackScore == -1 || score < fallbackScore {
+				fallbackScore = score
+				fallbackDiskId = diskId
+			}
 		}
 	}

-	return bestDiskId
+	// Return matching disk type if found, otherwise fallback
+	if bestDiskId != 0 {
+		return bestDiskId
+	}
+	return fallbackDiskId
 }

 // pickEcNodeAndDiskToBalanceShardsInto picks both a destination node and specific disk
@@ -1118,7 +1136,8 @@ func (ecb *ecBalancer) pickEcNodeAndDiskToBalanceShardsInto(vid needle.VolumeId,
 		return nil, 0, err
 	}

-	diskId := pickBestDiskOnNode(node, vid)
+	// For balancing, strictly require matching disk type
+	diskId := pickBestDiskOnNode(node, vid, ecb.diskType, true)
 	return node, diskId, nil
 }

@@ -1134,14 +1153,14 @@ func (ecb *ecBalancer) pickOneEcNodeAndMoveOneShard(existingLocation *EcNode, co
 	} else {
 		fmt.Printf("%s moves ec shard %d.%d to %s\n", existingLocation.info.Id, vid, shardId, destNode.info.Id)
 	}
-	return moveMountedShardToEcNode(ecb.commandEnv, existingLocation, collection, vid, shardId, destNode, destDiskId, ecb.applyBalancing)
+	return moveMountedShardToEcNode(ecb.commandEnv, existingLocation, collection, vid, shardId, destNode, destDiskId, ecb.applyBalancing, ecb.diskType)
 }

-func pickNEcShardsToMoveFrom(ecNodes []*EcNode, vid needle.VolumeId, n int) map[erasure_coding.ShardId]*EcNode {
+func pickNEcShardsToMoveFrom(ecNodes []*EcNode, vid needle.VolumeId, n int, diskType types.DiskType) map[erasure_coding.ShardId]*EcNode {
 	picked := make(map[erasure_coding.ShardId]*EcNode)
 	var candidateEcNodes []*CandidateEcNode
 	for _, ecNode := range ecNodes {
-		shardBits := findEcVolumeShards(ecNode, vid)
+		shardBits := findEcVolumeShards(ecNode, vid, diskType)
 		if shardBits.ShardIdCount() > 0 {
 			candidateEcNodes = append(candidateEcNodes, &CandidateEcNode{
 				ecNode:     ecNode,
@@ -1155,13 +1174,13 @@ func pickNEcShardsToMoveFrom(ecNodes []*EcNode, vid needle.VolumeId, n int) map[
 	for i := 0; i < n; i++ {
 		selectedEcNodeIndex := -1
 		for i, candidateEcNode := range candidateEcNodes {
-			shardBits := findEcVolumeShards(candidateEcNode.ecNode, vid)
+			shardBits := findEcVolumeShards(candidateEcNode.ecNode, vid, diskType)
 			if shardBits > 0 {
 				selectedEcNodeIndex = i
 				for _, shardId := range shardBits.ShardIds() {
 					candidateEcNode.shardCount--
 					picked[shardId] = candidateEcNode.ecNode
-					candidateEcNode.ecNode.deleteEcVolumeShards(vid, []uint32{uint32(shardId)})
+					candidateEcNode.ecNode.deleteEcVolumeShards(vid, []uint32{uint32(shardId)}, diskType)
 					break
 				}
 				break
@@ -1180,7 +1199,7 @@ func pickNEcShardsToMoveFrom(ecNodes []*EcNode, vid needle.VolumeId, n int) map[
 func (ecb *ecBalancer) collectVolumeIdToEcNodes(collection string) map[needle.VolumeId][]*EcNode {
 	vidLocations := make(map[needle.VolumeId][]*EcNode)
 	for _, ecNode := range ecb.ecNodes {
-		diskInfo, found := ecNode.info.DiskInfos[string(types.HardDriveType)]
+		diskInfo, found := ecNode.info.DiskInfos[string(ecb.diskType)]
 		if !found {
 			continue
 		}
@@ -1194,9 +1213,9 @@ func (ecb *ecBalancer) collectVolumeIdToEcNodes(collection string) map[needle.Vo
 	return vidLocations
 }

-func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, maxParallelization int, applyBalancing bool) (err error) {
+func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool) (err error) {
 	// collect all ec nodes
-	allEcNodes, totalFreeEcSlots, err := collectEcNodesForDC(commandEnv, dc)
+	allEcNodes, totalFreeEcSlots, err := collectEcNodesForDC(commandEnv, dc, diskType)
 	if err != nil {
 		return err
 	}
@@ -1210,6 +1229,7 @@ func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplic
 		replicaPlacement:   ecReplicaPlacement,
 		applyBalancing:     applyBalancing,
 		maxParallelization: maxParallelization,
+		diskType:           diskType,
 	}

 	if len(collections) == 0 {
@@ -106,7 +106,7 @@ func TestParseReplicaPlacementArg(t *testing.T) {
 func TestEcDistribution(t *testing.T) {

 	// find out all volume servers with one slot left.
-	ecNodes, totalFreeEcSlots := collectEcVolumeServersByDc(topology1, "")
+	ecNodes, totalFreeEcSlots := collectEcVolumeServersByDc(topology1, "", types.HardDriveType)

 	sortEcNodesByFreeslotsDescending(ecNodes)

@@ -149,16 +149,17 @@ func TestPickRackToBalanceShardsInto(t *testing.T) {

 	for _, tc := range testCases {
 		vid, _ := needle.NewVolumeId(tc.vid)
-		ecNodes, _ := collectEcVolumeServersByDc(tc.topology, "")
+		ecNodes, _ := collectEcVolumeServersByDc(tc.topology, "", types.HardDriveType)
 		rp, _ := super_block.NewReplicaPlacementFromString(tc.replicaPlacement)

 		ecb := &ecBalancer{
 			ecNodes:          ecNodes,
 			replicaPlacement: rp,
+			diskType:         types.HardDriveType,
 		}

 		racks := ecb.racks()
-		rackToShardCount := countShardsByRack(vid, ecNodes)
+		rackToShardCount := countShardsByRack(vid, ecNodes, types.HardDriveType)

 		got, gotErr := ecb.pickRackToBalanceShardsInto(racks, rackToShardCount)
 		if err := errorCheck(gotErr, tc.wantErr); err != nil {
@@ -225,10 +226,11 @@ func TestPickEcNodeToBalanceShardsInto(t *testing.T) {

 	for _, tc := range testCases {
 		vid, _ := needle.NewVolumeId(tc.vid)
-		allEcNodes, _ := collectEcVolumeServersByDc(tc.topology, "")
+		allEcNodes, _ := collectEcVolumeServersByDc(tc.topology, "", types.HardDriveType)

 		ecb := &ecBalancer{
-			ecNodes: allEcNodes,
+			ecNodes:  allEcNodes,
+			diskType: types.HardDriveType,
 		}

 		// Resolve target node by name
@@ -32,13 +32,23 @@ func (c *commandEcDecode) Name() string {
 func (c *commandEcDecode) Help() string {
 	return `decode a erasure coded volume into a normal volume

-	ec.decode [-collection=""] [-volumeId=<volume_id>]
+	ec.decode [-collection=""] [-volumeId=<volume_id>] [-diskType=<disk_type>]

 	The -collection parameter supports regular expressions for pattern matching:
 	  - Use exact match: ec.decode -collection="^mybucket$"
 	  - Match multiple buckets: ec.decode -collection="bucket.*"
 	  - Match all collections: ec.decode -collection=".*"

+	Options:
+	  -diskType: source disk type where EC shards are stored (hdd, ssd, or empty for default hdd)
+
+	Examples:
+	  # Decode EC shards from HDD (default)
+	  ec.decode -collection=mybucket
+
+	  # Decode EC shards from SSD
+	  ec.decode -collection=mybucket -diskType=ssd
+
 `
 }

@@ -50,6 +60,7 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 	decodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := decodeCommand.Int("volumeId", 0, "the volume id")
 	collection := decodeCommand.String("collection", "", "the collection name")
+	diskTypeStr := decodeCommand.String("diskType", "", "source disk type where EC shards are stored (hdd, ssd, or empty for default hdd)")
 	if err = decodeCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -59,6 +70,7 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 	}

 	vid := needle.VolumeId(*volumeId)
+	diskType := types.ToDiskType(*diskTypeStr)

 	// collect topology information
 	topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
@@ -68,17 +80,17 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr

 	// volumeId is provided
 	if vid != 0 {
-		return doEcDecode(commandEnv, topologyInfo, *collection, vid)
+		return doEcDecode(commandEnv, topologyInfo, *collection, vid, diskType)
 	}

 	// apply to all volumes in the collection
-	volumeIds, err := collectEcShardIds(topologyInfo, *collection)
+	volumeIds, err := collectEcShardIds(topologyInfo, *collection, diskType)
 	if err != nil {
 		return err
 	}
 	fmt.Printf("ec decode volumes: %v\n", volumeIds)
 	for _, vid := range volumeIds {
-		if err = doEcDecode(commandEnv, topologyInfo, *collection, vid); err != nil {
+		if err = doEcDecode(commandEnv, topologyInfo, *collection, vid, diskType); err != nil {
 			return err
 		}
 	}
@@ -86,14 +98,14 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 	return nil
 }

-func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collection string, vid needle.VolumeId) (err error) {
+func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collection string, vid needle.VolumeId, diskType types.DiskType) (err error) {

 	if !commandEnv.isLocked() {
 		return fmt.Errorf("lock is lost")
 	}

 	// find volume location
-	nodeToEcIndexBits := collectEcNodeShardBits(topoInfo, vid)
+	nodeToEcIndexBits := collectEcNodeShardBits(topoInfo, vid, diskType)

 	fmt.Printf("ec volume %d shard locations: %+v\n", vid, nodeToEcIndexBits)

@@ -248,7 +260,7 @@ func lookupVolumeIds(commandEnv *CommandEnv, volumeIds []string) (volumeIdLocati
 	return resp.VolumeIdLocations, nil
 }

-func collectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionPattern string) (vids []needle.VolumeId, err error) {
+func collectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionPattern string, diskType types.DiskType) (vids []needle.VolumeId, err error) {
 	// compile regex pattern for collection matching
 	collectionRegex, err := compileCollectionPattern(collectionPattern)
 	if err != nil {
@@ -257,7 +269,7 @@ func collectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionPattern strin

 	vidMap := make(map[uint32]bool)
 	eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
-		if diskInfo, found := dn.DiskInfos[string(types.HardDriveType)]; found {
+		if diskInfo, found := dn.DiskInfos[string(diskType)]; found {
 			for _, v := range diskInfo.EcShardInfos {
 				if collectionRegex.MatchString(v.Collection) {
 					vidMap[v.Id] = true
@@ -273,11 +285,11 @@ func collectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionPattern strin
 	return
 }

-func collectEcNodeShardBits(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId) map[pb.ServerAddress]erasure_coding.ShardBits {
+func collectEcNodeShardBits(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, diskType types.DiskType) map[pb.ServerAddress]erasure_coding.ShardBits {

 	nodeToEcIndexBits := make(map[pb.ServerAddress]erasure_coding.ShardBits)
 	eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
-		if diskInfo, found := dn.DiskInfos[string(types.HardDriveType)]; found {
+		if diskInfo, found := dn.DiskInfos[string(diskType)]; found {
 			for _, v := range diskInfo.EcShardInfos {
 				if v.Id == uint32(vid) {
 					nodeToEcIndexBits[pb.NewServerAddressFromDataNode(dn)] = erasure_coding.ShardBits(v.EcIndexBits)
@@ -37,8 +37,8 @@ func (c *commandEcEncode) Name() string {
 func (c *commandEcEncode) Help() string {
 	return `apply erasure coding to a volume

-	ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h] [-verbose]
-	ec.encode [-collection=""] [-volumeId=<volume_id>] [-verbose]
+	ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h] [-verbose] [-sourceDiskType=<disk_type>] [-diskType=<disk_type>]
+	ec.encode [-collection=""] [-volumeId=<volume_id>] [-verbose] [-diskType=<disk_type>]

 	This command will:
 	1. freeze one volume
@@ -61,6 +61,18 @@ func (c *commandEcEncode) Help() string {

 	Options:
 	  -verbose: show detailed reasons why volumes are not selected for encoding
+	  -sourceDiskType: filter source volumes by disk type (hdd, ssd, or empty for all)
+	  -diskType: target disk type for EC shards (hdd, ssd, or empty for default hdd)
+
+	Examples:
+	  # Encode SSD volumes to SSD EC shards (same tier)
+	  ec.encode -collection=mybucket -sourceDiskType=ssd -diskType=ssd
+
+	  # Encode SSD volumes to HDD EC shards (tier migration to cheaper storage)
+	  ec.encode -collection=mybucket -sourceDiskType=ssd -diskType=hdd
+
+	  # Encode all volumes to SSD EC shards
+	  ec.encode -collection=mybucket -diskType=ssd

 	Re-balancing algorithm:
 	` + ecBalanceAlgorithmDescription
@@ -80,6 +92,8 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 	maxParallelization := encodeCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
 	forceChanges := encodeCommand.Bool("force", false, "force the encoding even if the cluster has less than recommended 4 nodes")
 	shardReplicaPlacement := encodeCommand.String("shardReplicaPlacement", "", "replica placement for EC shards, or master default if empty")
+	sourceDiskTypeStr := encodeCommand.String("sourceDiskType", "", "filter source volumes by disk type (hdd, ssd, or empty for all)")
+	diskTypeStr := encodeCommand.String("diskType", "", "target disk type for EC shards (hdd, ssd, or empty for default hdd)")
 	applyBalancing := encodeCommand.Bool("rebalance", false, "re-balance EC shards after creation")
 	verbose := encodeCommand.Bool("verbose", false, "show detailed reasons why volumes are not selected for encoding")

@@ -94,6 +108,16 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 		return err
 	}

+	// Parse source disk type filter (optional)
+	var sourceDiskType *types.DiskType
+	if *sourceDiskTypeStr != "" {
+		sdt := types.ToDiskType(*sourceDiskTypeStr)
+		sourceDiskType = &sdt
+	}
+
+	// Parse target disk type for EC shards
+	diskType := types.ToDiskType(*diskTypeStr)
+
 	// collect topology information
 	topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
 	if err != nil {
@@ -119,7 +143,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 		balanceCollections = collectCollectionsForVolumeIds(topologyInfo, volumeIds)
 	} else {
 		// apply to all volumes for the given collection pattern (regex)
-		volumeIds, balanceCollections, err = collectVolumeIdsForEcEncode(commandEnv, *collection, nil, *fullPercentage, *quietPeriod, *verbose)
+		volumeIds, balanceCollections, err = collectVolumeIdsForEcEncode(commandEnv, *collection, sourceDiskType, *fullPercentage, *quietPeriod, *verbose)
 		if err != nil {
 			return err
 		}
@@ -142,7 +166,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 		return fmt.Errorf("ec encode for volumes %v: %w", volumeIds, err)
 	}
 	// ...re-balance ec shards...
-	if err := EcBalance(commandEnv, balanceCollections, "", rp, *maxParallelization, *applyBalancing); err != nil {
+	if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, *maxParallelization, *applyBalancing); err != nil {
 		return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err)
 	}
 	// ...then delete original volumes using pre-collected locations.
@@ -12,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )

 func init() {
@@ -24,6 +25,7 @@ type ecRebuilder struct {
 	writer       io.Writer
 	applyChanges bool
 	collections  []string
+	diskType     types.DiskType

 	ewg       *ErrorWaitGroup
 	ecNodesMu sync.Mutex
@@ -39,7 +41,7 @@ func (c *commandEcRebuild) Name() string {
 func (c *commandEcRebuild) Help() string {
 	return `find and rebuild missing ec shards among volume servers

-	ec.rebuild [-c EACH_COLLECTION|<collection_name>] [-apply] [-maxParallelization N]
+	ec.rebuild [-c EACH_COLLECTION|<collection_name>] [-apply] [-maxParallelization N] [-diskType=<disk_type>]

 	Options:
 	  -collection: specify a collection name, or "EACH_COLLECTION" to process all collections
@@ -47,6 +49,7 @@ func (c *commandEcRebuild) Help() string {
 	  -maxParallelization: number of volumes to rebuild concurrently (default: 10)
 	                       Increase for faster rebuilds with more system resources.
 	                       Decrease if experiencing resource contention or instability.
+	  -diskType: disk type for EC shards (hdd, ssd, or empty for default hdd)

 	Algorithm:

@@ -83,6 +86,7 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W
 	collection := fixCommand.String("collection", "EACH_COLLECTION", "collection name, or \"EACH_COLLECTION\" for each collection")
 	maxParallelization := fixCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
 	applyChanges := fixCommand.Bool("apply", false, "apply the changes")
+	diskTypeStr := fixCommand.String("diskType", "", "disk type for EC shards (hdd, ssd, or empty for default hdd)")
 	// TODO: remove this alias
 	applyChangesAlias := fixCommand.Bool("force", false, "apply the changes (alias for -apply)")
 	if err = fixCommand.Parse(args); err != nil {
@@ -95,8 +99,10 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W
 		return
 	}

+	diskType := types.ToDiskType(*diskTypeStr)
+
 	// collect all ec nodes
-	allEcNodes, _, err := collectEcNodes(commandEnv)
+	allEcNodes, _, err := collectEcNodes(commandEnv, diskType)
 	if err != nil {
 		return err
 	}
@@ -117,6 +123,7 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W
 		writer:       writer,
 		applyChanges: *applyChanges,
 		collections:  collections,
+		diskType:     diskType,

 		ewg: NewErrorWaitGroup(*maxParallelization),
 	}
@@ -294,7 +301,7 @@ func (erb *ecRebuilder) rebuildOneEcVolume(collection string, volumeId needle.Vo
 	// ensure ECNode updates are atomic
 	erb.ecNodesMu.Lock()
 	defer erb.ecNodesMu.Unlock()
-	rebuilder.addEcVolumeShards(volumeId, collection, generatedShardIds)
+	rebuilder.addEcVolumeShards(volumeId, collection, generatedShardIds, erb.diskType)

 	return nil
 }
@@ -5,6 +5,7 @@ import (

 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )

 func TestCommandEcBalanceSmall(t *testing.T) {
@@ -14,6 +15,7 @@ func TestCommandEcBalanceSmall(t *testing.T) {
 			newEcNode("dc1", "rack2", "dn2", 100).addEcVolumeAndShardsForTest(2, "c1", []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13}),
 		},
 		applyBalancing: false,
+		diskType:       types.HardDriveType,
 	}

 	ecb.balanceEcVolumes("c1")
@@ -30,6 +32,7 @@ func TestCommandEcBalanceNothingToMove(t *testing.T) {
 				addEcVolumeAndShardsForTest(2, "c1", []uint32{0, 1, 2, 3, 4, 5, 6}),
 		},
 		applyBalancing: false,
+		diskType:       types.HardDriveType,
 	}

 	ecb.balanceEcVolumes("c1")
@@ -48,6 +51,7 @@ func TestCommandEcBalanceAddNewServers(t *testing.T) {
 			newEcNode("dc1", "rack1", "dn4", 100),
 		},
 		applyBalancing: false,
+		diskType:       types.HardDriveType,
 	}

 	ecb.balanceEcVolumes("c1")
@@ -66,6 +70,7 @@ func TestCommandEcBalanceAddNewRacks(t *testing.T) {
 			newEcNode("dc1", "rack2", "dn4", 100),
 		},
 		applyBalancing: false,
+		diskType:       types.HardDriveType,
 	}

 	ecb.balanceEcVolumes("c1")
@@ -109,6 +114,7 @@ func TestCommandEcBalanceVolumeEvenButRackUneven(t *testing.T) {
 			newEcNode("dc1", "rack1", "dn3", 100),
 		},
 		applyBalancing: false,
+		diskType:       types.HardDriveType,
 	}

 	ecb.balanceEcVolumes("c1")
@@ -128,5 +134,5 @@ func newEcNode(dc string, rack string, dataNodeId string, freeEcSlot int) *EcNod
 }

 func (ecNode *EcNode) addEcVolumeAndShardsForTest(vid uint32, collection string, shardIds []uint32) *EcNode {
-	return ecNode.addEcVolumeShards(needle.VolumeId(vid), collection, shardIds)
+	return ecNode.addEcVolumeShards(needle.VolumeId(vid), collection, shardIds, types.HardDriveType)
 }
@@ -156,21 +156,30 @@ func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandE
 }

 func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
-	// find this ec volume server
+	// Evacuate EC volumes for all disk types discovered from topology
+	// Disk types are free-form tags (e.g., "", "hdd", "ssd", "nvme", etc.)
+	// We need to handle each disk type separately because shards should be moved to nodes with the same disk type
 	// We collect topology once at the start and track capacity changes ourselves
 	// (via freeEcSlot decrement after each move) rather than repeatedly refreshing,
 	// which would give a false sense of correctness since topology could be stale.
-	ecNodes, _ := collectEcVolumeServersByDc(c.topologyInfo, "")
-	thisNodes, otherNodes := c.ecNodesOtherThan(ecNodes, volumeServer)
-	if len(thisNodes) == 0 {
-		return fmt.Errorf("%s is not found in this cluster\n", volumeServer)
-	}
+	diskTypes := collectVolumeDiskTypes(c.topologyInfo)

-	// move away ec volumes
-	for _, thisNode := range thisNodes {
-		for _, diskInfo := range thisNode.info.DiskInfos {
+	for _, diskType := range diskTypes {
+		ecNodes, _ := collectEcVolumeServersByDc(c.topologyInfo, "", diskType)
+		thisNodes, otherNodes := c.ecNodesOtherThan(ecNodes, volumeServer)
+		if len(thisNodes) == 0 {
+			// This server doesn't have EC shards for this disk type, skip
+			continue
+		}
+
+		// move away ec volumes for this disk type
+		for _, thisNode := range thisNodes {
+			diskInfo, found := thisNode.info.DiskInfos[string(diskType)]
+			if !found {
+				continue
+			}
 			for _, ecShardInfo := range diskInfo.EcShardInfos {
-				hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange, writer)
+				hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange, diskType, writer)
 				if err != nil {
 					fmt.Fprintf(writer, "move away volume %d from %s: %v\n", ecShardInfo.Id, volumeServer, err)
 				}
@@ -187,7 +196,7 @@ func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv,
 	return nil
 }

-func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv, ecShardInfo *master_pb.VolumeEcShardInformationMessage, thisNode *EcNode, otherNodes []*EcNode, applyChange bool, writer io.Writer) (hasMoved bool, err error) {
+func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv, ecShardInfo *master_pb.VolumeEcShardInformationMessage, thisNode *EcNode, otherNodes []*EcNode, applyChange bool, diskType types.DiskType, writer io.Writer) (hasMoved bool, err error) {

 	for _, shardId := range erasure_coding.ShardBits(ecShardInfo.EcIndexBits).ShardIds() {
 		// Sort by: 1) fewest shards of this volume, 2) most free EC slots
@@ -217,13 +226,14 @@ func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv
 				collectionPrefix = ecShardInfo.Collection + "_"
 			}
 			vid := needle.VolumeId(ecShardInfo.Id)
-			destDiskId := pickBestDiskOnNode(emptyNode, vid)
+			// For evacuation, prefer same disk type but allow fallback to other types
+			destDiskId := pickBestDiskOnNode(emptyNode, vid, diskType, false)
 			if destDiskId > 0 {
 				fmt.Fprintf(writer, "moving ec volume %s%d.%d %s => %s (disk %d)\n", collectionPrefix, ecShardInfo.Id, shardId, thisNode.info.Id, emptyNode.info.Id, destDiskId)
 			} else {
 				fmt.Fprintf(writer, "moving ec volume %s%d.%d %s => %s\n", collectionPrefix, ecShardInfo.Id, shardId, thisNode.info.Id, emptyNode.info.Id)
 			}
-			err = moveMountedShardToEcNode(commandEnv, thisNode, ecShardInfo.Collection, vid, shardId, emptyNode, destDiskId, applyChange)
+			err = moveMountedShardToEcNode(commandEnv, thisNode, ecShardInfo.Collection, vid, shardId, emptyNode, destDiskId, applyChange, diskType)
 			if err != nil {
 				hasMoved = false
 				return
@@ -55,6 +55,7 @@ func (n *Needle) ReadNeedleMeta(r backend.BackendStorageFile, offset int64, size
 		if OffsetSize == 4 && offset < int64(MaxPossibleVolumeSize) {
 			return ErrorSizeMismatch
 		}
+		return fmt.Errorf("size mismatch for entry at offset %d: found id %x size %d, expected size %d", offset, n.Id, n.Size, size)
 	}
 	n.DataSize = util.BytesToUint32(bytes[NeedleHeaderSize : NeedleHeaderSize+DataSizeSize])
 	startOffset := offset + NeedleHeaderSize
@@ -42,6 +42,14 @@ func MaybeLoadVolumeInfo(fileName string) (volumeInfo *volume_server_pb.VolumeIn

 	}

+	// Handle empty .vif files gracefully - treat as if file doesn't exist
+	// This can happen when ec.decode copies from a source that doesn't have a .vif file
+	if len(fileData) == 0 {
+		glog.Warningf("empty volume info file %s, treating as non-existent", fileName)
+		hasVolumeInfoFile = false
+		return
+	}
+
 	glog.V(1).Infof("maybeLoadVolumeInfo Unmarshal volume info %v", fileName)
 	if err = jsonpb.Unmarshal(fileData, volumeInfo); err != nil {
 		if oldVersionErr := tryOldVersionVolumeInfo(fileData, volumeInfo); oldVersionErr != nil {
@@ -723,7 +723,10 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		}
 	}
 	if tsMemory.IsZero() { // case 2.2
-		return nil, -2, nil
+		// Buffer is empty - return ResumeFromDiskError so caller can read from disk
+		// This fixes issue #4977 where SubscribeMetadata stalls because
+		// MetaAggregator.MetaLogBuffer is empty in single-filer setups
+		return nil, -2, ResumeFromDiskError
 	} else if lastReadPosition.Time.Before(tsMemory) { // case 2.3
 		// For time-based reads, only check timestamp for disk reads
 		// Don't use offset comparisons as they're not meaningful for time-based subscriptions