refactor: remove unused gRPC connection age parameters

The GrpcMaxConnectionAge and GrpcMaxConnectionAgeGrace constants have a troubled history - they were removed in 2022 due to gRPC issues, reverted later, and recently re-added. However, they are not essential to the core worker reconnection fix which was solved through proper goroutine ordering. The Docker Swarm DNS handling mentioned in the comments is not critical, and these parameters have caused problems in the past. Removing them simplifies the configuration without losing functionality.
refactor: extract drain and close channel logic into helper function
2026-10-17 12:05:52 +00:00 · 2025-12-22 18:19:52 -08:00 · 2025-12-22 17:56:42 -08:00 · 2025-12-22 17:55:43 -08:00 · 2025-12-22 17:55:36 -08:00 · 2025-12-22 17:55:25 -08:00
26 changed files with 2014 additions and 282 deletions
@@ -18,7 +18,7 @@ defaults:
 jobs:
  awscli-tests:
    runs-on: ubuntu-latest
-    timeout-minutes: 5
+    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v6

@@ -96,12 +96,12 @@ require (
 	gocloud.dev v0.43.0
 	gocloud.dev/pubsub/natspubsub v0.43.0
 	gocloud.dev/pubsub/rabbitpubsub v0.43.0
-	golang.org/x/crypto v0.45.0
+	golang.org/x/crypto v0.46.0
 	golang.org/x/exp v0.0.0-20250811191247-51f88131bc50
 	golang.org/x/image v0.34.0
-	golang.org/x/net v0.47.0
+	golang.org/x/net v0.48.0
 	golang.org/x/oauth2 v0.32.0 // indirect
-	golang.org/x/sys v0.38.0
+	golang.org/x/sys v0.39.0
 	golang.org/x/text v0.32.0 // indirect
 	golang.org/x/tools v0.39.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
@@ -127,8 +127,8 @@ require (
 	github.com/arangodb/go-driver v1.6.9
 	github.com/armon/go-metrics v0.4.1
 	github.com/aws/aws-sdk-go-v2 v1.41.0
-	github.com/aws/aws-sdk-go-v2/config v1.31.3
-	github.com/aws/aws-sdk-go-v2/credentials v1.19.5
+	github.com/aws/aws-sdk-go-v2/config v1.32.6
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.6
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0
 	github.com/cognusion/imaging v1.0.2
 	github.com/fluent/fluent-logger-golang v1.10.1
@@ -158,7 +158,7 @@ require (
 	github.com/tarantool/go-tarantool/v2 v2.4.1
 	github.com/tikv/client-go/v2 v2.0.7
 	github.com/xeipuuv/gojsonschema v1.2.0
-	github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.0
+	github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.1
 	github.com/ydb-platform/ydb-go-sdk/v3 v3.122.0
 	go.etcd.io/etcd/client/pkg/v3 v3.6.6
 	go.uber.org/atomic v1.11.0
@@ -173,6 +173,7 @@ require (
 	cloud.google.com/go/longrunning v0.6.7 // indirect
 	cloud.google.com/go/pubsub/v2 v2.0.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.7.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 // indirect
 	github.com/bazelbuild/rules_go v0.46.0 // indirect
 	github.com/biogo/store v0.0.0-20201120204734-aad293a2328f // indirect
 	github.com/blevesearch/snowballstem v0.9.0 // indirect
@@ -264,7 +265,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.6 // indirect
@@ -272,7 +273,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sns v1.34.7 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sqs v1.38.8 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect
 	github.com/aws/smithy-go v1.24.0 // indirect
@@ -445,7 +446,7 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/arch v0.20.0 // indirect
-	golang.org/x/term v0.37.0 // indirect
+	golang.org/x/term v0.38.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20251022142026-3a174f9686a8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect
@@ -669,10 +669,10 @@ github.com/aws/aws-sdk-go-v2 v1.41.0 h1:tNvqh1s+v0vFYdA1xq0aOJH+Y5cRyZ5upu6roPgP
 github.com/aws/aws-sdk-go-v2 v1.41.0/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4/go.mod h1:IOAPF6oT9KCsceNTvvYMNHy0+kMF8akOjeDvPENWxp4=
-github.com/aws/aws-sdk-go-v2/config v1.31.3 h1:RIb3yr/+PZ18YYNe6MDiG/3jVoJrPmdoCARwNkMGvco=
-github.com/aws/aws-sdk-go-v2/config v1.31.3/go.mod h1:jjgx1n7x0FAKl6TnakqrpkHWWKcX3xfWtdnIJs5K9CE=
-github.com/aws/aws-sdk-go-v2/credentials v1.19.5 h1:xMo63RlqP3ZZydpJDMBsH9uJ10hgHYfQFIk1cHDXrR4=
-github.com/aws/aws-sdk-go-v2/credentials v1.19.5/go.mod h1:hhbH6oRcou+LpXfA/0vPElh/e0M3aFeOblE1sssAAEk=
+github.com/aws/aws-sdk-go-v2/config v1.32.6 h1:hFLBGUKjmLAekvi1evLi5hVvFQtSo3GYwi+Bx4lpJf8=
+github.com/aws/aws-sdk-go-v2/config v1.32.6/go.mod h1:lcUL/gcd8WyjCrMnxez5OXkO3/rwcNmvfno62tnXNcI=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.6 h1:F9vWao2TwjV2MyiyVS+duza0NIRtAslgLUM0vTA1ZaE=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.6/go.mod h1:SgHzKjEVsdQr6Opor0ihgWtkWdfRAIwxYzSJ8O85VHY=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 h1:80+uETIWS1BqjnN9uJ0dBUaETh+P1XwFy5vwHwK5r9k=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16/go.mod h1:wOOsYuxYuB/7FlnVtzeBYRcjSRtQpAW0hCP7tIULMwo=
 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4 h1:0SzCLoPRSK3qSydsaFQWugP+lOBCTPwfcBOm6222+UA=
@@ -681,8 +681,8 @@ github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 h1:rgGwPzb82iBYSvHMH
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16/go.mod h1:L/UxsGeKpGoIj6DxfhOWHWQ/kGKcd4I1VncE4++IyKA=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 h1:1jtGzuV7c82xnqOVfx2F0xmJcOw5374L7N6juGW6x6U=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16/go.mod h1:M2E5OQf+XLe+SZGmmpaI2yy+J326aFf6/+54PoxSANc=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 h1:WKuaxf++XKWlHWu9ECbMlha8WOEGm0OUEZqm4K/Gcfk=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4/go.mod h1:ZWy7j6v1vWGmPReu0iSGvRiise4YI5SkR3OHKTZ6Wuc=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15 h1:NLYTEyZmVZo0Qh183sC8nC+ydJXOOeIL/qI/sS3PdLY=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15/go.mod h1:Z803iB3B0bc8oJV8zH2PERLRfQUJ2n2BXISpsA4+O1M=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 h1:0ryTNEdJbzUCEWkVXEXoqlXV72J5keC1GvILMOuD00E=
@@ -695,12 +695,14 @@ github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15 h1:wsSQ4SVz5YE1c
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15/go.mod h1:I7sditnFGtYMIqPRU1QoHZAUrXkGp4SczmlLwrNPlD0=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0 h1:IrbE3B8O9pm3lsg96AXIN5MXX4pECEuExh/A0Du3AuI=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0/go.mod h1:/sJLzHtiiZvs6C1RbxS/anSAFwZD6oC6M/kotQzOiLw=
+github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 h1:HpI7aMmJ+mm1wkSHIA2t5EaFFv5EFYXePW30p1EIrbQ=
+github.com/aws/aws-sdk-go-v2/service/signin v1.0.4/go.mod h1:C5RdGMYGlfM0gYq/tifqgn4EbyX99V15P2V3R+VHbQU=
 github.com/aws/aws-sdk-go-v2/service/sns v1.34.7 h1:OBuZE9Wt8h2imuRktu+WfjiTGrnYdCIJg8IX92aalHE=
 github.com/aws/aws-sdk-go-v2/service/sns v1.34.7/go.mod h1:4WYoZAhHt+dWYpoOQUgkUKfuQbE6Gg/hW4oXE0pKS9U=
 github.com/aws/aws-sdk-go-v2/service/sqs v1.38.8 h1:80dpSqWMwx2dAm30Ib7J6ucz1ZHfiv5OCRwN/EnCOXQ=
 github.com/aws/aws-sdk-go-v2/service/sqs v1.38.8/go.mod h1:IzNt/udsXlETCdvBOL0nmyMe2t9cGmXmZgsdoZGYYhI=
-github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 h1:eYnlt6QxnFINKzwxP5/Ucs1vkG7VT3Iezmvfgc2waUw=
-github.com/aws/aws-sdk-go-v2/service/sso v1.30.7/go.mod h1:+fWt2UHSb4kS7Pu8y+BMBvJF0EWx+4H0hzNwtDNRTrg=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 h1:aM/Q24rIlS3bRAhTyFurowU8A0SMyGDtEOY/l/s/1Uw=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.8/go.mod h1:+fWt2UHSb4kS7Pu8y+BMBvJF0EWx+4H0hzNwtDNRTrg=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 h1:AHDr0DaHIAo8c9t1emrzAlVDFp+iMMKnPdYy6XO4MCE=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12/go.mod h1:GQ73XawFFiWxyWXMHWfhiomvP3tXtdNar/fi8z18sx0=
 github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 h1:SciGFVNZ4mHdm7gpD1dgZYnCuVdX1s+lFTg4+4DOy70=
@@ -1777,8 +1779,8 @@ github.com/ydb-platform/ydb-go-genproto v0.0.0-20221215182650-986f9d10542f/go.mo
 github.com/ydb-platform/ydb-go-genproto v0.0.0-20230528143953-42c825ace222/go.mod h1:Er+FePu1dNUieD+XTMDduGpQuCPssK5Q4BjF+IIXJ3I=
 github.com/ydb-platform/ydb-go-genproto v0.0.0-20251125145508-6d7ef87db5cb h1:LZ6dhVfWzhicf/P5Xh7fA0Jd7rfGduxmB2QZpD+Lz9Q=
 github.com/ydb-platform/ydb-go-genproto v0.0.0-20251125145508-6d7ef87db5cb/go.mod h1:Er+FePu1dNUieD+XTMDduGpQuCPssK5Q4BjF+IIXJ3I=
-github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.0 h1:/NyPd9KnCJgzrEXCArqk1ThqCH2Dh31uUwl88o/VkuM=
-github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.0/go.mod h1:9YzkhlIymWaJGX6KMU3vh5sOf3UKbCXkG/ZdjaI3zNM=
+github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.1 h1:XaRxeVrOyl3y6v9CiYMWaFdZ6zevvYe+TRxOR8ifa2s=
+github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.1/go.mod h1:9YzkhlIymWaJGX6KMU3vh5sOf3UKbCXkG/ZdjaI3zNM=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.44.0/go.mod h1:oSLwnuilwIpaF5bJJMAofnGgzPJusoI3zWMNb8I+GnM=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.47.3/go.mod h1:bWnOIcUHd7+Sl7DN+yhyY1H/I61z53GczvwJgXMgvj0=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.122.0 h1:uvqheUfoEJz0CLLLaYct69wpvoaosM/2joWFcbwoqfw=
@@ -1917,8 +1919,8 @@ golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+
 golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
-golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
-golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
+golang.org/x/crypto v0.46.0 h1:cKRW/pmt1pKAfetfu+RCEvjvZkA9RimPbh7bhFjGVBU=
+golang.org/x/crypto v0.46.0/go.mod h1:Evb/oLKmMraqjZ2iQTwDwvCtJkczlDuTmdJXoZVzqU0=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -2056,8 +2058,8 @@ golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
-golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
-golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
+golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU=
+golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -2218,8 +2220,8 @@ golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
-golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
+golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -2237,8 +2239,8 @@ golang.org/x/term v0.19.0/go.mod h1:2CuTdWZ7KHSQwUzKva0cbMg6q2DMI3Mmxp+gKJbskEk=
 golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
 golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
-golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU=
-golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254=
+golang.org/x/term v0.38.0 h1:PQ5pkm/rLO6HnxFR7N2lJHOZX6Kez5Y1gDSJla6jo7Q=
+golang.org/x/term v0.38.0/go.mod h1:bSEAKrOT1W+VSu9TSCMtoGEOUcKxOKgl3LE5QEF/xVg=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -43,24 +43,24 @@ require (
 	github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc // indirect
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/aws/aws-sdk-go v1.55.8 // indirect
-	github.com/aws/aws-sdk-go-v2 v1.40.1 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.41.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
 	github.com/aws/aws-sdk-go-v2/config v1.31.3 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.19.3 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.15 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.5 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.15 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.15 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.6 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.15 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.16 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15 // indirect
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.30.6 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.11 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.41.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect
 	github.com/aws/smithy-go v1.24.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bradenaw/juniper v0.15.3 // indirect
@@ -229,14 +229,14 @@ require (
 	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/crypto v0.45.0 // indirect
+	golang.org/x/crypto v0.46.0 // indirect
 	golang.org/x/exp v0.0.0-20250811191247-51f88131bc50 // indirect
 	golang.org/x/image v0.34.0 // indirect
-	golang.org/x/net v0.47.0 // indirect
+	golang.org/x/net v0.48.0 // indirect
 	golang.org/x/oauth2 v0.32.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect
-	golang.org/x/sys v0.38.0 // indirect
-	golang.org/x/term v0.37.0 // indirect
+	golang.org/x/sys v0.39.0 // indirect
+	golang.org/x/term v0.38.0 // indirect
 	golang.org/x/text v0.32.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
 	google.golang.org/api v0.247.0 // indirect
@@ -102,22 +102,22 @@ github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3d
 github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
 github.com/aws/aws-sdk-go v1.55.8 h1:JRmEUbU52aJQZ2AjX4q4Wu7t4uZjOu71uyNmaWlUkJQ=
 github.com/aws/aws-sdk-go v1.55.8/go.mod h1:ZkViS9AqA6otK+JBBNH2++sx1sgxrPKcSzPPvQkUtXk=
-github.com/aws/aws-sdk-go-v2 v1.40.1 h1:difXb4maDZkRH0x//Qkwcfpdg1XQVXEAEs2DdXldFFc=
-github.com/aws/aws-sdk-go-v2 v1.40.1/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0=
+github.com/aws/aws-sdk-go-v2 v1.41.0 h1:tNvqh1s+v0vFYdA1xq0aOJH+Y5cRyZ5upu6roPgPKd4=
+github.com/aws/aws-sdk-go-v2 v1.41.0/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4/go.mod h1:IOAPF6oT9KCsceNTvvYMNHy0+kMF8akOjeDvPENWxp4=
 github.com/aws/aws-sdk-go-v2/config v1.31.3 h1:RIb3yr/+PZ18YYNe6MDiG/3jVoJrPmdoCARwNkMGvco=
 github.com/aws/aws-sdk-go-v2/config v1.31.3/go.mod h1:jjgx1n7x0FAKl6TnakqrpkHWWKcX3xfWtdnIJs5K9CE=
-github.com/aws/aws-sdk-go-v2/credentials v1.19.3 h1:01Ym72hK43hjwDeJUfi1l2oYLXBAOR8gNSZNmXmvuas=
-github.com/aws/aws-sdk-go-v2/credentials v1.19.3/go.mod h1:55nWF/Sr9Zvls0bGnWkRxUdhzKqj9uRNlPvgV1vgxKc=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.15 h1:utxLraaifrSBkeyII9mIbVwXXWrZdlPO7FIKmyLCEcY=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.15/go.mod h1:hW6zjYUDQwfz3icf4g2O41PHi77u10oAzJ84iSzR/lo=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.5 h1:xMo63RlqP3ZZydpJDMBsH9uJ10hgHYfQFIk1cHDXrR4=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.5/go.mod h1:hhbH6oRcou+LpXfA/0vPElh/e0M3aFeOblE1sssAAEk=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 h1:80+uETIWS1BqjnN9uJ0dBUaETh+P1XwFy5vwHwK5r9k=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16/go.mod h1:wOOsYuxYuB/7FlnVtzeBYRcjSRtQpAW0hCP7tIULMwo=
 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4 h1:0SzCLoPRSK3qSydsaFQWugP+lOBCTPwfcBOm6222+UA=
 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.18.4/go.mod h1:JAet9FsBHjfdI+TnMBX4ModNNaQHAd3dc/Bk+cNsxeM=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.15 h1:Y5YXgygXwDI5P4RkteB5yF7v35neH7LfJKBG+hzIons=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.15/go.mod h1:K+/1EpG42dFSY7CBj+Fruzm8PsCGWTXJ3jdeJ659oGQ=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.15 h1:AvltKnW9ewxX2hFmQS0FyJH93aSvJVUEFvXfU+HWtSE=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.15/go.mod h1:3I4oCdZdmgrREhU74qS1dK9yZ62yumob+58AbFR4cQA=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 h1:rgGwPzb82iBYSvHMHXc8h9mRoOUBZIGFgKb9qniaZZc=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16/go.mod h1:L/UxsGeKpGoIj6DxfhOWHWQ/kGKcd4I1VncE4++IyKA=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16 h1:1jtGzuV7c82xnqOVfx2F0xmJcOw5374L7N6juGW6x6U=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.16/go.mod h1:M2E5OQf+XLe+SZGmmpaI2yy+J326aFf6/+54PoxSANc=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.15 h1:NLYTEyZmVZo0Qh183sC8nC+ydJXOOeIL/qI/sS3PdLY=
@@ -126,18 +126,18 @@ github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 h1:0ryTNEd
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4/go.mod h1:HQ4qwNZh32C3CBeO6iJLQlgtMzqeG17ziAA/3KDJFow=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.6 h1:P1MU/SuhadGvg2jtviDXPEejU3jBNhoeeAlRadHzvHI=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.6/go.mod h1:5KYaMG6wmVKMFBSfWoyG/zH8pWwzQFnKgpoSRlXHKdQ=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.15 h1:3/u/4yZOffg5jdNk1sDpOQ4Y+R6Xbh+GzpDrSZjuy3U=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.15/go.mod h1:4Zkjq0FKjE78NKjabuM4tRXKFzUJWXgP0ItEZK8l7JU=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.16 h1:oHjJHeUy0ImIV0bsrX0X91GkV5nJAyv1l1CC9lnO0TI=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.16/go.mod h1:iRSNGgOYmiYwSCXxXaKb9HfOEj40+oTKn8pTxMlYkRM=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15 h1:wsSQ4SVz5YE1crz0Ap7VBZrV4nNqZt4CIBBT8mnwoNc=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.15/go.mod h1:I7sditnFGtYMIqPRU1QoHZAUrXkGp4SczmlLwrNPlD0=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0 h1:IrbE3B8O9pm3lsg96AXIN5MXX4pECEuExh/A0Du3AuI=
 github.com/aws/aws-sdk-go-v2/service/s3 v1.93.0/go.mod h1:/sJLzHtiiZvs6C1RbxS/anSAFwZD6oC6M/kotQzOiLw=
-github.com/aws/aws-sdk-go-v2/service/sso v1.30.6 h1:8sTTiw+9yuNXcfWeqKF2x01GqCF49CpP4Z9nKrrk/ts=
-github.com/aws/aws-sdk-go-v2/service/sso v1.30.6/go.mod h1:8WYg+Y40Sn3X2hioaaWAAIngndR8n1XFdRPPX+7QBaM=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.11 h1:E+KqWoVsSrj1tJ6I/fjDIu5xoS2Zacuu1zT+H7KtiIk=
-github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.11/go.mod h1:qyWHz+4lvkXcr3+PoGlGHEI+3DLLiU6/GdrFfMaAhB0=
-github.com/aws/aws-sdk-go-v2/service/sts v1.41.3 h1:tzMkjh0yTChUqJDgGkcDdxvZDSrJ/WB6R6ymI5ehqJI=
-github.com/aws/aws-sdk-go-v2/service/sts v1.41.3/go.mod h1:T270C0R5sZNLbWUe8ueiAF42XSZxxPocTaGSgs5c/60=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 h1:eYnlt6QxnFINKzwxP5/Ucs1vkG7VT3Iezmvfgc2waUw=
+github.com/aws/aws-sdk-go-v2/service/sso v1.30.7/go.mod h1:+fWt2UHSb4kS7Pu8y+BMBvJF0EWx+4H0hzNwtDNRTrg=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 h1:AHDr0DaHIAo8c9t1emrzAlVDFp+iMMKnPdYy6XO4MCE=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12/go.mod h1:GQ73XawFFiWxyWXMHWfhiomvP3tXtdNar/fi8z18sx0=
+github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 h1:SciGFVNZ4mHdm7gpD1dgZYnCuVdX1s+lFTg4+4DOy70=
+github.com/aws/aws-sdk-go-v2/service/sts v1.41.5/go.mod h1:iW40X4QBmUxdP+fZNOpfmkdMZqsovezbAeO+Ubiv2pk=
 github.com/aws/smithy-go v1.24.0 h1:LpilSUItNPFr1eY85RYgTIg5eIEPtvFbskaFcmmIUnk=
 github.com/aws/smithy-go v1.24.0/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -730,8 +730,8 @@ golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+
 golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
-golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
-golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
+golang.org/x/crypto v0.46.0 h1:cKRW/pmt1pKAfetfu+RCEvjvZkA9RimPbh7bhFjGVBU=
+golang.org/x/crypto v0.46.0/go.mod h1:Evb/oLKmMraqjZ2iQTwDwvCtJkczlDuTmdJXoZVzqU0=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -811,8 +811,8 @@ golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
-golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
-golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
+golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU=
+golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -890,8 +890,8 @@ golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
-golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
+golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -904,8 +904,8 @@ golang.org/x/term v0.19.0/go.mod h1:2CuTdWZ7KHSQwUzKva0cbMg6q2DMI3Mmxp+gKJbskEk=
 golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
 golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
-golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU=
-golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254=
+golang.org/x/term v0.38.0 h1:PQ5pkm/rLO6HnxFR7N2lJHOZX6Kez5Y1gDSJla6jo7Q=
+golang.org/x/term v0.38.0/go.mod h1:bSEAKrOT1W+VSu9TSCMtoGEOUcKxOKgl3LE5QEF/xVg=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -3,6 +3,7 @@ package cluster
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync/atomic"
 	"time"

@@ -41,6 +42,7 @@ type LiveLock struct {
 	self           string
 	lc             *LockClient
 	owner          string
+	lockTTL        time.Duration
 }

 // NewShortLivedLock creates a lock with a 5-second duration
@@ -59,17 +61,24 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc
 }

 // StartLongLivedLock starts a goroutine to lock the key and returns immediately.
-func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string)) (lock *LiveLock) {
+// lockTTL specifies how long the lock should be held. The renewal interval is
+// automatically derived as lockTTL / 2 to ensure timely renewals.
+func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string), lockTTL time.Duration) (lock *LiveLock) {
 	lock = &LiveLock{
 		key:            key,
 		hostFiler:      lc.seedFiler,
 		cancelCh:       make(chan struct{}),
-		expireAtNs:     time.Now().Add(lock_manager.LiveLockTTL).UnixNano(),
+		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,
 		self:           owner,
 		lc:             lc,
+		lockTTL:        lockTTL,
+	}
+	if lock.lockTTL == 0 {
+		lock.lockTTL = lock_manager.LiveLockTTL
 	}
 	go func() {
+		renewInterval := lock.lockTTL / 2
 		isLocked := false
 		lockOwner := ""
 		for {
@@ -82,13 +91,13 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh
 			}

 			if isLocked {
-				if err := lock.AttemptToLock(lock_manager.LiveLockTTL); err != nil {
+				if err := lock.AttemptToLock(lock.lockTTL); err != nil {
 					glog.V(0).Infof("Lost lock %s: %v", key, err)
 					isLocked = false
 					atomic.StoreInt32(&lock.isLocked, 0)
 				}
 			} else {
-				if err := lock.AttemptToLock(lock_manager.LiveLockTTL); err == nil {
+				if err := lock.AttemptToLock(lock.lockTTL); err == nil {
 					isLocked = true
 					// Note: AttemptToLock already sets lock.isLocked atomically on success
 				}
@@ -102,7 +111,11 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh
 			case <-lock.cancelCh:
 				return
 			default:
-				time.Sleep(lock_manager.RenewInterval)
+				if isLocked {
+					time.Sleep(renewInterval)
+				} else {
+					time.Sleep(5 * renewInterval)
+				}
 			}
 		}
 	}()
@@ -129,7 +142,11 @@ func (lock *LiveLock) AttemptToLock(lockDuration time.Duration) error {
 		return err
 	}
 	if errorMessage != "" {
-		glog.V(1).Infof("LOCK: doLock returned error message for key=%s: %s", lock.key, errorMessage)
+		if strings.Contains(errorMessage, "lock already owned") {
+			glog.V(3).Infof("LOCK: doLock returned error message for key=%s: %s", lock.key, errorMessage)
+		} else {
+			glog.V(2).Infof("LOCK: doLock returned error message for key=%s: %s", lock.key, errorMessage)
+		}
 		time.Sleep(time.Second)
 		return fmt.Errorf("%v", errorMessage)
 	}
@@ -206,7 +223,7 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e
 			errorMessage = resp.Error
 			if resp.LockHostMovedTo != "" && resp.LockHostMovedTo != string(previousHostFiler) {
 				// Only log if the host actually changed
-				glog.V(1).Infof("LOCK: Host changed from %s to %s for key=%s", previousHostFiler, resp.LockHostMovedTo, lock.key)
+				glog.V(2).Infof("LOCK: Host changed from %s to %s for key=%s", previousHostFiler, resp.LockHostMovedTo, lock.key)
 				lock.hostFiler = pb.ServerAddress(resp.LockHostMovedTo)
 				lock.lc.seedFiler = lock.hostFiler
 			} else if resp.LockHostMovedTo != "" {
@@ -214,12 +231,12 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e
 			}
 			if resp.LockOwner != "" && resp.LockOwner != previousOwner {
 				// Only log if the owner actually changed
-				glog.V(1).Infof("LOCK: Owner changed from %s to %s for key=%s", previousOwner, resp.LockOwner, lock.key)
+				glog.V(2).Infof("LOCK: Owner changed from %s to %s for key=%s", previousOwner, resp.LockOwner, lock.key)
 				lock.owner = resp.LockOwner
 			} else if resp.LockOwner != "" {
 				lock.owner = resp.LockOwner
 			} else if previousOwner != "" {
-				glog.V(1).Infof("LOCK: Owner cleared for key=%s", lock.key)
+				glog.V(2).Infof("LOCK: Owner cleared for key=%s", lock.key)
 				lock.owner = ""
 			}
 		}
@@ -32,6 +32,7 @@ var Commands = []*Command{
 	cmdIam,
 	cmdMaster,
 	cmdMasterFollower,
+	cmdMini,
 	cmdMount,
 	cmdMqAgent,
 	cmdMqBroker,
@@ -246,7 +246,7 @@ func (s3opt *S3Options) startS3Server() bool {
 			return nil
 		})
 		if err != nil {
-			glog.V(0).Infof("wait to connect to filers %v grpc address", filerAddresses)
+			glog.V(2).Infof("wait to connect to filers %v grpc address", filerAddresses)
 			time.Sleep(time.Second)
 		} else {
 			glog.V(0).Infof("connected to filers %v", filerAddresses)
@@ -3,13 +3,14 @@ package command
 import (
 	"context"
 	"fmt"
-	"github.com/seaweedfs/seaweedfs/weed/util/version"
 	"net/http"
 	"os"
 	"os/user"
 	"strconv"
 	"time"

+	"github.com/seaweedfs/seaweedfs/weed/util/version"
+
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -102,7 +103,7 @@ func (wo *WebDavOption) startWebDav() bool {
 			return nil
 		})
 		if err != nil {
-			glog.V(0).Infof("wait to connect to filer %s grpc address %s", *wo.filer, filerAddress.ToGrpcAddress())
+			glog.V(2).Infof("wait to connect to filer %s grpc address %s", *wo.filer, filerAddress.ToGrpcAddress())
 			time.Sleep(time.Second)
 		} else {
 			glog.V(0).Infof("connected to filer %s grpc address %s", *wo.filer, filerAddress.ToGrpcAddress())
@@ -12,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"

 	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 	"google.golang.org/grpc"
@@ -142,7 +143,7 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
 		mqBroker.lockAsBalancer = lockClient.StartLongLivedLock(pub_balancer.LockBrokerBalancer, string(self), func(newLockOwner string) {
 			glog.V(0).Infof("broker %s found balanacer %s", self, newLockOwner)
 			newBrokerBalancerCh <- newLockOwner
-		})
+		}, lock_manager.LiveLockTTL)
 		mqBroker.KeepConnectedToBrokerBalancer(newBrokerBalancerCh)
 	}()

@@ -12,6 +12,7 @@ import (
 	"time"

 	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/filer_client"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -180,6 +181,7 @@ func (cr *CoordinatorRegistry) startLeaderElection() {
 			GatewayLeaderLockKey,
 			cr.gatewayAddress,
 			cr.onLeadershipChange,
+			lock_manager.LiveLockTTL,
 		)

 		// Wait for shutdown
@@ -35,11 +35,6 @@ const (
 	// gRPC keepalive settings - must be consistent between client and server
 	GrpcKeepAliveTime    = 60 * time.Second // ping interval when no activity
 	GrpcKeepAliveTimeout = 20 * time.Second // ping timeout
-
-	// Connection recycling for Docker Swarm environments
-	// Forces connections to be recycled periodically to handle DNS changes
-	GrpcMaxConnectionAge      = 5 * time.Minute  // max time a connection may exist
-	GrpcMaxConnectionAgeGrace = 30 * time.Second // grace period for RPCs to complete
 )

 var (
@@ -63,10 +58,8 @@ func NewGrpcServer(opts ...grpc.ServerOption) *grpc.Server {
 	var options []grpc.ServerOption
 	options = append(options,
 		grpc.KeepaliveParams(keepalive.ServerParameters{
-			Time:                  GrpcKeepAliveTime,         // server pings client if no activity for this long
-			Timeout:               GrpcKeepAliveTimeout,      // ping timeout
-			MaxConnectionAge:      GrpcMaxConnectionAge,      // max connection age for Docker Swarm DNS refresh
-			MaxConnectionAgeGrace: GrpcMaxConnectionAgeGrace, // grace period for in-flight RPCs
+			Time:    GrpcKeepAliveTime,    // server pings client if no activity for this long
+			Timeout: GrpcKeepAliveTimeout, // ping timeout
 		}),
 		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
 			MinTime:             GrpcKeepAliveTime, // min time a client should wait before sending a ping
@@ -2,12 +2,13 @@ package filersink

 import (
 	"fmt"
-	"github.com/schollz/progressbar/v3"
-	"github.com/seaweedfs/seaweedfs/weed/util"
 	"os"
 	"path/filepath"
 	"sync"

+	"github.com/schollz/progressbar/v3"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+
 	"google.golang.org/grpc"

 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -114,6 +115,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string)
 			IsInputCompressed: "gzip" == header.Get("Content-Encoding"),
 			MimeType:          header.Get("Content-Type"),
 			PairMap:           nil,
+			RetryForever:      true,
 		},
 		func(host, fileId string) string {
 			fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
@@ -61,7 +61,7 @@ func (s3a *S3ApiServer) startBucketSizeMetricsLoop(ctx context.Context) {
 	// Start long-lived lock - this S3 instance will only collect metrics when it holds the lock
 	lock := lockClient.StartLongLivedLock(s3MetricsLockName, owner, func(newLockOwner string) {
 		glog.V(1).Infof("S3 bucket size metrics lock owner changed to: %s", newLockOwner)
-	})
+	}, bucketSizeMetricsInterval)
 	defer lock.Stop()

 	ticker := time.NewTicker(bucketSizeMetricsInterval)
@@ -16,47 +16,32 @@ func TestImplicitDirectoryBehaviorLogic(t *testing.T) {
 		hasTrailingSlash  bool
 		fileSize          uint64
 		isDirectory       bool
-		mimeType          string
 		hasChildren       bool
 		versioningEnabled bool
 		shouldReturn404   bool
 		description       string
 	}{
 		{
-			name:              "PyArrow directory marker: 0-byte file with application/octet-stream and children",
+			name:              "Implicit directory: 0-byte file with children, no trailing slash",
 			objectPath:        "dataset",
 			hasTrailingSlash:  false,
 			fileSize:          0,
 			isDirectory:       false,
-			mimeType:          "application/octet-stream",
 			hasChildren:       true,
 			versioningEnabled: false,
 			shouldReturn404:   true,
 			description:       "Should return 404 to force s3fs LIST-based discovery",
 		},
 		{
-			name:              "PyArrow directory marker: 0-byte file with empty MIME type and children",
-			objectPath:        "dataset",
-			hasTrailingSlash:  false,
-			fileSize:          0,
-			isDirectory:       false,
-			mimeType:          "",
-			hasChildren:       true,
-			versioningEnabled: false,
-			shouldReturn404:   true,
-			description:       "Should return 404 for empty MIME type directory markers",
-		},
-		{
-			name:              "Actual directory with children",
+			name:              "Implicit directory: actual directory with children, no trailing slash",
 			objectPath:        "dataset",
 			hasTrailingSlash:  false,
 			fileSize:          0,
 			isDirectory:       true,
-			mimeType:          "",
 			hasChildren:       true,
 			versioningEnabled: false,
-			shouldReturn404:   false,
-			description:       "Should return 200 for actual directories (maintains AWS S3 compatibility)",
+			shouldReturn404:   true,
+			description:       "Should return 404 for directory with children",
 		},
 		{
 			name:              "Explicit directory request: trailing slash",
@@ -64,7 +49,6 @@ func TestImplicitDirectoryBehaviorLogic(t *testing.T) {
 			hasTrailingSlash:  true,
 			fileSize:          0,
 			isDirectory:       true,
-			mimeType:          "",
 			hasChildren:       true,
 			versioningEnabled: false,
 			shouldReturn404:   false,
@@ -76,19 +60,17 @@ func TestImplicitDirectoryBehaviorLogic(t *testing.T) {
 			hasTrailingSlash:  false,
 			fileSize:          0,
 			isDirectory:       false,
-			mimeType:          "application/octet-stream",
 			hasChildren:       false,
 			versioningEnabled: false,
 			shouldReturn404:   false,
 			description:       "Should return 200 for legitimate empty file",
 		},
 		{
-			name:              "Empty directory: directory without children",
+			name:              "Empty directory: 0-byte directory without children",
 			objectPath:        "empty-dir",
 			hasTrailingSlash:  false,
 			fileSize:          0,
 			isDirectory:       true,
-			mimeType:          "",
 			hasChildren:       false,
 			versioningEnabled: false,
 			shouldReturn404:   false,
@@ -100,44 +82,41 @@ func TestImplicitDirectoryBehaviorLogic(t *testing.T) {
 			hasTrailingSlash:  false,
 			fileSize:          100,
 			isDirectory:       false,
-			mimeType:          "text/plain",
 			hasChildren:       false,
 			versioningEnabled: false,
 			shouldReturn404:   false,
 			description:       "Should return 200 for regular file with content",
 		},
 		{
-			name:              "Versioned bucket: directory marker should return 200",
+			name:              "Versioned bucket: implicit directory should return 200",
 			objectPath:        "dataset",
 			hasTrailingSlash:  false,
 			fileSize:          0,
 			isDirectory:       false,
-			mimeType:          "application/octet-stream",
 			hasChildren:       true,
 			versioningEnabled: true,
 			shouldReturn404:   false,
-			description:       "Should return 200 for versioned buckets (skip directory marker check)",
+			description:       "Should return 200 for versioned buckets (skip implicit dir check)",
 		},
 		{
-			name:              "Directory marker with specific MIME type",
+			name:              "PyArrow directory marker: 0-byte with children",
 			objectPath:        "dataset",
 			hasTrailingSlash:  false,
 			fileSize:          0,
 			isDirectory:       false,
-			mimeType:          "text/plain",
 			hasChildren:       true,
 			versioningEnabled: false,
-			shouldReturn404:   false,
-			description:       "Should return 200 for 0-byte files with specific MIME types (not generic markers)",
+			shouldReturn404:   true,
+			description:       "Should return 404 for PyArrow-created directory markers",
 		},
 	}

 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// Test the logic: should we return 404?
-			// New logic from HeadObjectHandler:
+			// Logic from HeadObjectHandler:
 			// if !versioningConfigured && !strings.HasSuffix(object, "/") {
-			//     if isZeroByteFile && hasGenericMimeType {
+			//     if isZeroByteFile || isActualDirectory {
 			//         if hasChildren {
 			//             return 404
 			//         }
@@ -145,11 +124,11 @@ func TestImplicitDirectoryBehaviorLogic(t *testing.T) {
 			// }

 			isZeroByteFile := tt.fileSize == 0 && !tt.isDirectory
-			hasGenericMimeType := tt.mimeType == "" || tt.mimeType == "application/octet-stream"
+			isActualDirectory := tt.isDirectory

 			shouldReturn404 := false
 			if !tt.versioningEnabled && !tt.hasTrailingSlash {
-				if isZeroByteFile && hasGenericMimeType {
+				if isZeroByteFile || isActualDirectory {
 					if tt.hasChildren {
 						shouldReturn404 = true
 					}
@@ -249,18 +228,18 @@ func TestImplicitDirectoryEdgeCases(t *testing.T) {
 		expectation string
 	}{
 		{
-			name:        "PyArrow write_dataset creates 0-byte files with application/octet-stream",
-			scenario:    "PyArrow creates 'dataset' as 0-byte file with MIME type 'application/octet-stream', then writes 'dataset/file.parquet'",
-			expectation: "HEAD dataset → 404 (has children + generic MIME type), s3fs uses LIST → correctly identifies as directory",
+			name:        "PyArrow write_dataset creates 0-byte files",
+			scenario:    "PyArrow creates 'dataset' as 0-byte file, then writes 'dataset/file.parquet'",
+			expectation: "HEAD dataset → 404 (has children), s3fs uses LIST → correctly identifies as directory",
 		},
 		{
 			name:        "Filer creates actual directories",
 			scenario:    "Filer creates 'dataset' as actual directory with IsDirectory=true",
-			expectation: "HEAD dataset → 200 (actual directory, not 0-byte file), maintains AWS S3 compatibility",
+			expectation: "HEAD dataset → 404 (has children), s3fs uses LIST → correctly identifies as directory",
 		},
 		{
 			name:        "Empty file edge case",
-			scenario:    "User creates 'empty.txt' as 0-byte file with 'application/octet-stream' but no children",
+			scenario:    "User creates 'empty.txt' as 0-byte file with no children",
 			expectation: "HEAD empty.txt → 200 (no children), s3fs correctly reports as file",
 		},
 		{
@@ -271,18 +250,13 @@ func TestImplicitDirectoryEdgeCases(t *testing.T) {
 		{
 			name:        "Versioned bucket",
 			scenario:    "Bucket has versioning enabled",
-			expectation: "HEAD dataset → 200 (skip directory marker check), versioned semantics apply",
+			expectation: "HEAD dataset → 200 (skip implicit dir check), versioned semantics apply",
 		},
 		{
 			name:        "AWS S3 compatibility",
 			scenario:    "Only 'dataset/file.txt' exists, no marker at 'dataset'",
 			expectation: "HEAD dataset → 404 (object doesn't exist), matches AWS S3 behavior",
 		},
-		{
-			name:        "Directory marker with specific MIME type",
-			scenario:    "PyArrow creates 'dataset' as 0-byte file with MIME type 'text/plain' and children",
-			expectation: "HEAD dataset → 200 (specific MIME type, not generic), may not work with PyArrow but preserves compatibility",
-		},
 	}

 	for _, tt := range tests {
@@ -2278,45 +2278,47 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 	//
 	// Background:
 	//   Some S3 clients (like PyArrow with s3fs) create directory markers when writing datasets.
-	//   These are typically 0-byte files with MIME type "application/octet-stream" that have children.
+	//   These can be either:
+	//   1. 0-byte files with directory MIME type (e.g., "application/octet-stream")
+	//   2. Actual directories in the filer (created by PyArrow's write_dataset)
 	//
 	// Problem:
-	//   s3fs's info() method calls HEAD on these markers. If HEAD returns 200 with size=0,
-	//   s3fs incorrectly reports them as files instead of directories, causing PyArrow to fail.
+	//   s3fs's info() method calls HEAD on the path. If HEAD returns 200 with size=0,
+	//   s3fs incorrectly reports it as a file (type='file', size=0) instead of checking
+	//   for children. This causes PyArrow to fail with "Parquet file size is 0 bytes".
 	//
 	// Solution:
 	//   For non-versioned objects without trailing slash, if the object is a 0-byte file
-	//   with MIME type "application/octet-stream" AND has children, return 404 instead of 200.
-	//   This forces s3fs to fall back to LIST-based discovery, which correctly identifies
-	//   the object as a directory.
+	//   or directory AND has children, return 404 instead of 200. This forces s3fs to
+	//   fall back to LIST-based discovery, which correctly identifies it as a directory.
 	//
 	// AWS S3 Compatibility:
-	//   This maintains AWS S3 compatibility by only affecting objects that are likely
-	//   directory markers (0-byte files with generic MIME type that have children).
-	//   Regular 0-byte files with "application/octet-stream" that are not directory markers
-	//   will still return 200 OK.
+	//   AWS S3 typically doesn't create directory markers for implicit directories, so
+	//   HEAD on "dataset" (when only "dataset/file.txt" exists) returns 404. Our behavior
+	//   matches this by returning 404 for implicit directories with children.
 	//
 	// Edge Cases Handled:
-	//   - Regular empty files (0-byte, no children) → 200 OK
-	//   - Empty directories (no children) → 200 OK
+	//   - Empty files (0-byte, no children) → 200 OK (legitimate empty file)
+	//   - Empty directories (no children) → 200 OK (legitimate empty directory)
 	//   - Explicit directory requests (trailing slash) → 200 OK (handled earlier)
 	//   - Versioned objects → Skip this check (different semantics)
-	//   - Directory markers with other MIME types → 200 OK (may break PyArrow but preserves compatibility)
 	//
 	// Performance:
-	//   Only adds overhead for 0-byte files with "application/octet-stream" MIME type.
+	//   Only adds overhead for 0-byte files or directories without trailing slash.
 	//   Cost: One LIST operation with Limit=1 (~1-5ms).
 	//
 	if !versioningConfigured && !strings.HasSuffix(object, "/") {
-		// Check if this looks like a PyArrow directory marker
+		// Check if this is an implicit directory (either a 0-byte file or actual directory with children)
+		// PyArrow may create 0-byte files when writing datasets, or the filer may have actual directories
 		if objectEntryForSSE.Attributes != nil {
 			isZeroByteFile := objectEntryForSSE.Attributes.FileSize == 0 && !objectEntryForSSE.IsDirectory
-			hasGenericMimeType := objectEntryForSSE.Attributes.Mime == "" || objectEntryForSSE.Attributes.Mime == "application/octet-stream"
+			isActualDirectory := objectEntryForSSE.IsDirectory

-			if isZeroByteFile && hasGenericMimeType {
-				// Check if it has children (likely a directory marker)
+			if isZeroByteFile || isActualDirectory {
+				// Check if it has children (making it an implicit directory)
 				if s3a.hasChildren(bucket, object) {
-					// This appears to be a directory marker - return 404 to force LIST-based discovery
+					// This is an implicit directory with children
+					// Return 404 to force clients (like s3fs) to use LIST-based discovery
 					s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 					return
 				}
@@ -53,19 +53,16 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ

 	// collect parameters
 	bucket, _ := s3_constants.GetBucketAndObject(r)
-	glog.V(3).Infof("ListObjectsV2Handler %s", bucket)
-
 	originalPrefix, startAfter, delimiter, continuationToken, encodingTypeUrl, fetchOwner, maxKeys, allowUnordered, errCode := getListObjectsV2Args(r.URL.Query())

+	glog.V(2).Infof("ListObjectsV2Handler bucket=%s prefix=%s", bucket, originalPrefix)
+
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
 		return
 	}

-	if maxKeys < 0 {
-		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxKeys)
-		return
-	}
+	// maxKeys is uint16 here; negative values are rejected during parsing.

 	// AWS S3 compatibility: allow-unordered cannot be used with delimiter
 	if allowUnordered && delimiter != "" {
@@ -121,10 +118,10 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ

 	// collect parameters
 	bucket, _ := s3_constants.GetBucketAndObject(r)
-	glog.V(3).Infof("ListObjectsV1Handler %s", bucket)
-
 	originalPrefix, marker, delimiter, encodingTypeUrl, maxKeys, allowUnordered, errCode := getListObjectsV1Args(r.URL.Query())

+	glog.V(2).Infof("ListObjectsV1Handler bucket=%s prefix=%s delimiter=%s maxKeys=%d", bucket, originalPrefix, delimiter, maxKeys)
+
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
 		return
@@ -206,7 +203,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
 		for {
 			empty := true

-			nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, cursor, marker, delimiter, false, func(dir string, entry *filer_pb.Entry) {
+			nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, cursor, marker, delimiter, false, bucket, func(dir string, entry *filer_pb.Entry) {
 				empty = false
 				dirName, entryName, _ := entryUrlEncode(dir, entry.Name, encodingTypeUrl)
 				if entry.IsDirectory {
@@ -310,6 +307,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
 						}
 					}
 					if !delimiterFound {
+						glog.V(4).Infof("Adding file to contents: %s", entryName)
 						contents = append(contents, newListEntry(entry, "", dirName, entryName, bucketPrefix, fetchOwner, false, false, s3a.iam))
 						cursor.maxKeys--
 						lastEntryWasCommonPrefix = false
@@ -442,14 +440,13 @@ func toParentAndDescendants(dirAndName string) (dir, name string) {
 	return
 }

-func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, cursor *ListingCursor, marker, delimiter string, inclusiveStartFrom bool, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
+func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, cursor *ListingCursor, marker, delimiter string, inclusiveStartFrom bool, bucket string, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
 	// invariants
 	//   prefix and marker should be under dir, marker may contain "/"
 	//   maxKeys should be updated for each recursion
 	// glog.V(4).Infof("doListFilerEntries dir: %s, prefix: %s, marker %s, maxKeys: %d, prefixEndsOnDelimiter: %+v", dir, prefix, marker, cursor.maxKeys, cursor.prefixEndsOnDelimiter)
-	if prefix == "/" && delimiter == "/" {
-		return
-	}
+	// When listing at bucket root with delimiter '/', prefix can be "/" after normalization.
+	// Returning early here would incorrectly hide all top-level entries (folders like "Veeam/").
 	if cursor.maxKeys <= 0 {
 		return // Don't set isTruncated here - let caller decide based on whether more entries exist
 	}
@@ -457,7 +454,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 	if strings.Contains(marker, "/") {
 		subDir, subMarker := toParentAndDescendants(marker)
 		// println("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker)
-		subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+subDir, "", cursor, subMarker, delimiter, false, eachEntryFn)
+		subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+subDir, "", cursor, subMarker, delimiter, false, bucket, eachEntryFn)
 		if subErr != nil {
 			err = subErr
 			return
@@ -490,10 +487,6 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 		return
 	}

-	// Track .versions directories found in this directory for later processing
-	// Store the full entry to avoid additional getEntry calls (N+1 query optimization)
-	var versionsDirs []*filer_pb.Entry
-
 	for {
 		resp, recvErr := stream.Recv()
 		if recvErr != nil {
@@ -508,7 +501,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d

 		if cursor.maxKeys <= 0 {
 			cursor.isTruncated = true
-			continue
+			break
 		}

 		// Set nextMarker only when we have quota to process this entry
@@ -528,24 +521,42 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 				continue
 			}

-			// Skip .versions directories in regular list operations but track them for logical object creation
-			// Store the full entry to avoid additional getEntry calls later
+			// Process .versions directories immediately to create logical versioned object entries
+			// These directories are never traversed (we continue here), so each is only encountered once
 			if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
-				glog.V(4).Infof("Found .versions directory: %s", entry.Name)
-				versionsDirs = append(versionsDirs, entry)
+				// Extract object name from .versions directory name
+				baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder)
+				// Construct full object path relative to bucket
+				bucketFullPath := s3a.option.BucketsPath + "/" + bucket
+				bucketRelativePath := strings.TrimPrefix(dir, bucketFullPath)
+				bucketRelativePath = strings.TrimPrefix(bucketRelativePath, "/")
+				var fullObjectPath string
+				if bucketRelativePath == "" {
+					fullObjectPath = baseObjectName
+				} else {
+					fullObjectPath = bucketRelativePath + "/" + baseObjectName
+				}
+				// Use metadata from the already-fetched .versions directory entry
+				if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, entry); err == nil {
+					eachEntryFn(dir, latestVersionEntry)
+				} else if !errors.Is(err, ErrDeleteMarker) {
+					// Log unexpected errors (delete markers are expected)
+					glog.V(2).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err)
+				}
 				continue
 			}

 			if delimiter != "/" || cursor.prefixEndsOnDelimiter {
+				// When delimiter is empty (recursive mode), recurse into directories but don't add them to results
+				// Only files and versioned objects should appear in results
 				if cursor.prefixEndsOnDelimiter {
 					cursor.prefixEndsOnDelimiter = false
 					if entry.IsDirectoryKeyObject() {
 						eachEntryFn(dir, entry)
 					}
-				} else {
-					eachEntryFn(dir, entry)
 				}
-				subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", cursor, "", delimiter, false, eachEntryFn)
+				// Recurse into subdirectory - don't add the directory itself to results
+				subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", cursor, "", delimiter, false, bucket, eachEntryFn)
 				if subErr != nil {
 					err = fmt.Errorf("doListFilerEntries2: %w", subErr)
 					return
@@ -568,57 +579,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 		}
 	}

-	// After processing all regular entries, handle versioned objects
-	// Create logical entries for objects that have .versions directories
-	// OPTIMIZATION: Use the already-fetched .versions directory entry to avoid N+1 queries
-	for _, versionsDir := range versionsDirs {
-		if cursor.maxKeys <= 0 {
-			cursor.isTruncated = true
-			break
-		}
-
-		// Update nextMarker to ensure pagination advances past this .versions directory
-		// This is critical to prevent infinite loops when results are truncated
-		nextMarker = versionsDir.Name
-
-		// Extract object name from .versions directory name (remove .versions suffix)
-		baseObjectName := strings.TrimSuffix(versionsDir.Name, s3_constants.VersionsFolder)
-
-		// Construct full object path relative to bucket
-		// dir is something like "/buckets/sea-test-1/Veeam/Backup/vbr/Config"
-		// we need to get the path relative to bucket: "Veeam/Backup/vbr/Config/Owner"
-		bucketPath := strings.TrimPrefix(dir, s3a.option.BucketsPath+"/")
-		bucketName := strings.Split(bucketPath, "/")[0]
-
-		// Remove bucket name from path to get directory within bucket
-		bucketRelativePath := strings.Join(strings.Split(bucketPath, "/")[1:], "/")
-
-		var fullObjectPath string
-		if bucketRelativePath == "" {
-			// Object is at bucket root
-			fullObjectPath = baseObjectName
-		} else {
-			// Object is in subdirectory
-			fullObjectPath = bucketRelativePath + "/" + baseObjectName
-		}
-
-		glog.V(4).Infof("Processing versioned object: baseObjectName=%s, bucketRelativePath=%s, fullObjectPath=%s",
-			baseObjectName, bucketRelativePath, fullObjectPath)
-
-		// OPTIMIZATION: Use metadata from the already-fetched .versions directory entry
-		// This avoids additional getEntry calls which cause high "find" usage
-		if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucketName, fullObjectPath, versionsDir); err == nil {
-			glog.V(4).Infof("Creating logical entry for versioned object: %s", fullObjectPath)
-			eachEntryFn(dir, latestVersionEntry)
-		} else if errors.Is(err, ErrDeleteMarker) {
-			// Expected: latest version is a delete marker, object should not appear in list
-			glog.V(4).Infof("Skipping versioned object %s: delete marker", fullObjectPath)
-		} else {
-			// Unexpected failure: missing metadata, fetch error, etc.
-			glog.V(3).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err)
-		}
-	}
-
+	// Versioned directories are processed above (lines 524-546)
 	return
 }

@@ -720,7 +681,6 @@ func (s3a *S3ApiServer) ensureDirectoryAllEmpty(filerClient filer_pb.SeaweedFile
 	return true, nil
 }

-
 // compareWithDelimiter compares two strings for sorting, treating the delimiter character
 // as having lower precedence than other characters to match AWS S3 behavior.
 // For example, with delimiter '/', 'foo/' should come before 'foo+1/' even though '+' < '/' in ASCII.
@@ -1,14 +1,63 @@
 package s3api

 import (
+	"context"
+	"io"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"

+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"github.com/stretchr/testify/assert"
+	grpc "google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 )

+type testListEntriesStream struct {
+	entries []*filer_pb.Entry
+	idx     int
+}
+
+func (s *testListEntriesStream) Recv() (*filer_pb.ListEntriesResponse, error) {
+	if s.idx >= len(s.entries) {
+		return nil, io.EOF
+	}
+	resp := &filer_pb.ListEntriesResponse{Entry: s.entries[s.idx]}
+	s.idx++
+	return resp, nil
+}
+
+func (s *testListEntriesStream) Header() (metadata.MD, error) { return metadata.MD{}, nil }
+func (s *testListEntriesStream) Trailer() metadata.MD         { return metadata.MD{} }
+func (s *testListEntriesStream) Close() error                 { return nil }
+func (s *testListEntriesStream) Context() context.Context     { return context.Background() }
+func (s *testListEntriesStream) SendMsg(m interface{}) error  { return nil }
+func (s *testListEntriesStream) RecvMsg(m interface{}) error  { return nil }
+func (s *testListEntriesStream) CloseSend() error             { return nil }
+
+type testFilerClient struct {
+	filer_pb.SeaweedFilerClient
+	entriesByDir map[string][]*filer_pb.Entry
+}
+
+func (c *testFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	entries := c.entriesByDir[in.Directory]
+	// Simplified mock: implements basic prefix filtering but ignores Limit, StartFromFileName, and InclusiveStartFrom
+	// to keep test logic focused. Prefix "/" is treated as no filter for bucket root compatibility.
+	if in.Prefix != "" && in.Prefix != "/" {
+		filtered := make([]*filer_pb.Entry, 0)
+		for _, e := range entries {
+			if strings.HasPrefix(e.Name, in.Prefix) {
+				filtered = append(filtered, e)
+			}
+		}
+		entries = filtered
+	}
+	return &testListEntriesStream{entries: entries}, nil
+}
+
 func TestListObjectsHandler(t *testing.T) {

 	// https://docs.aws.amazon.com/AmazonS3/latest/API/v2-RESTBucketGET.html
@@ -53,6 +102,13 @@ func Test_normalizePrefixMarker(t *testing.T) {
 		wantAlignedPrefix string
 		wantAlignedMarker string
 	}{
+		{"bucket root listing with delimiter",
+			args{"/",
+				""},
+			"",
+			"",
+			"",
+		},
 		{"prefix is a directory",
 			args{"/parentDir/data/",
 				""},
@@ -146,6 +202,31 @@ func TestAllowUnorderedParameterValidation(t *testing.T) {
 	})
 }

+func TestDoListFilerEntries_BucketRootPrefixSlashDelimiterSlash_ListsDirectories(t *testing.T) {
+	// Regression test for a bug where doListFilerEntries returned early when
+	// prefix == "/" && delimiter == "/", causing bucket-root folder listings
+	// (e.g. Veeam v13) to return empty results.
+
+	s3a := &S3ApiServer{}
+	client := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{Name: "Veeam", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+			},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1000}
+	seen := make([]string, 0)
+	_, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "/", cursor, "", "/", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		if entry.IsDirectory {
+			seen = append(seen, entry.Name)
+		}
+	})
+	assert.NoError(t, err)
+	assert.Contains(t, seen, "Veeam")
+}
+
 func TestAllowUnorderedWithDelimiterValidation(t *testing.T) {
 	t.Run("should return error when allow-unordered=true and delimiter are both present", func(t *testing.T) {
 		// Create a request with both allow-unordered=true and delimiter
@@ -0,0 +1,433 @@
+package s3api
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
+	grpc "google.golang.org/grpc"
+)
+
+// TestListObjectsWithVersionedObjects tests that versioned objects are properly listed
+// This validates the fix for duplicate versioned objects issue
+func TestListObjectsWithVersionedObjects(t *testing.T) {
+	now := time.Now().Unix()
+
+	// Create test filer client with versioned objects
+	filerClient := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				// Regular directory
+				{
+					Name:        "folder1",
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+				},
+				// .versions directory with metadata for versioned object
+				{
+					Name:        "file1.txt" + s3_constants.VersionsFolder,
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+					Extended: map[string][]byte{
+						s3_constants.ExtLatestVersionIdKey:    []byte("v1-abc123"),
+						s3_constants.ExtLatestVersionSizeKey:  []byte("1234"),
+						s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+						s3_constants.ExtLatestVersionETagKey:  []byte(fmt.Sprintf("\"%s\"", hex.EncodeToString([]byte("test-etag-1")))),
+					},
+				},
+				// Another .versions directory
+				{
+					Name:        "file2.txt" + s3_constants.VersionsFolder,
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+					Extended: map[string][]byte{
+						s3_constants.ExtLatestVersionIdKey:    []byte("v2-def456"),
+						s3_constants.ExtLatestVersionSizeKey:  []byte("5678"),
+						s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+						s3_constants.ExtLatestVersionETagKey:  []byte(fmt.Sprintf("\"%s\"", hex.EncodeToString([]byte("test-etag-2")))),
+					},
+				},
+			},
+			"/buckets/test-bucket/folder1": {
+				// Versioned object in subdirectory
+				{
+					Name:        "nested.txt" + s3_constants.VersionsFolder,
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+					Extended: map[string][]byte{
+						s3_constants.ExtLatestVersionIdKey:    []byte("v3-ghi789"),
+						s3_constants.ExtLatestVersionSizeKey:  []byte("9012"),
+						s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+						s3_constants.ExtLatestVersionETagKey:  []byte(fmt.Sprintf("\"%s\"", hex.EncodeToString([]byte("test-etag-3")))),
+					},
+				},
+			},
+		},
+	}
+
+	s3a := &S3ApiServer{
+		option: &S3ApiServerOption{
+			BucketsPath: "/buckets",
+		},
+	}
+
+	tests := []struct {
+		name             string
+		bucket           string
+		prefix           string
+		delimiter        string
+		expectedCount    int
+		expectedKeys     []string
+		expectedPrefixes []string
+	}{
+		{
+			name:          "List all objects including versioned (no delimiter)",
+			bucket:        "test-bucket",
+			prefix:        "",
+			delimiter:     "",
+			expectedCount: 3, // file1.txt, file2.txt, folder1/nested.txt
+			expectedKeys: []string{
+				"file1.txt",
+				"file2.txt",
+				"folder1/nested.txt",
+			},
+			expectedPrefixes: []string{},
+		},
+		{
+			name:          "List bucket root with delimiter",
+			bucket:        "test-bucket",
+			prefix:        "",
+			delimiter:     "/",
+			expectedCount: 2, // file1.txt, file2.txt (folder1/ becomes common prefix)
+			expectedKeys: []string{
+				"file1.txt",
+				"file2.txt",
+			},
+			expectedPrefixes: []string{
+				"folder1/",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Directly call doListFilerEntries with a callback to exercise the versioned objects
+			// listing logic. The callback mirrors the production listFilerEntries behavior for
+			// path extraction and accumulation so that this test validates the internal listing
+			// implementation in isolation from the HTTP layer.
+			cursor := &ListingCursor{maxKeys: uint16(tt.expectedCount + 10)}
+			contents := []ListEntry{}
+			commonPrefixes := []PrefixEntry{}
+			bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, tt.bucket)
+
+			_, err := s3a.doListFilerEntries(filerClient, bucketPrefix[:len(bucketPrefix)-1], tt.prefix, cursor, "", tt.delimiter, false, tt.bucket, func(dir string, entry *filer_pb.Entry) {
+				if cursor.maxKeys <= 0 {
+					return
+				}
+
+				if entry.IsDirectory {
+					if tt.delimiter == "/" {
+						// Extract relative path from bucket prefix
+						relDir := strings.TrimPrefix(dir, bucketPrefix[:len(bucketPrefix)-1])
+						if relDir != "" && relDir[0] == '/' {
+							relDir = relDir[1:]
+						}
+						prefix := relDir
+						if prefix != "" {
+							prefix += "/"
+						}
+						prefix += entry.Name + "/"
+
+						commonPrefixes = append(commonPrefixes, PrefixEntry{
+							Prefix: prefix,
+						})
+						cursor.maxKeys--
+					}
+				} else {
+					// Extract key from dir and entry name
+					relDir := strings.TrimPrefix(dir, bucketPrefix[:len(bucketPrefix)-1])
+					if relDir != "" && relDir[0] == '/' {
+						relDir = relDir[1:]
+					}
+					key := entry.Name
+					if relDir != "" {
+						key = relDir + "/" + entry.Name
+					}
+
+					contents = append(contents, ListEntry{
+						Key: key,
+					})
+					cursor.maxKeys--
+				}
+			})
+
+			assert.NoError(t, err, "doListFilerEntries should not return error")
+			assert.Equal(t, tt.expectedCount, len(contents), "Should return correct number of objects")
+			assert.Equal(t, len(tt.expectedPrefixes), len(commonPrefixes), "Should return correct number of common prefixes")
+
+			// Verify keys
+			actualKeys := make([]string, len(contents))
+			for i, entry := range contents {
+				actualKeys[i] = entry.Key
+			}
+			assert.ElementsMatch(t, tt.expectedKeys, actualKeys, "Should return expected keys")
+
+			// Verify common prefixes
+			actualPrefixes := make([]string, len(commonPrefixes))
+			for i, prefix := range commonPrefixes {
+				actualPrefixes[i] = prefix.Prefix
+			}
+			assert.ElementsMatch(t, tt.expectedPrefixes, actualPrefixes, "Should return expected prefixes")
+
+			// Verify each versioned object has correct version metadata
+			for _, entry := range contents {
+				assert.NotEmpty(t, entry.Key, "Versioned object should have key")
+			}
+		})
+	}
+}
+
+// TestVersionedObjectsNoDuplication ensures that .versions directories are only processed once
+// This is the core test for the bug fix - previously versioned objects were duplicated 250x
+func TestVersionedObjectsNoDuplication(t *testing.T) {
+	now := time.Now().Unix()
+
+	// Create a single .versions directory
+	filerClient := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{
+					Name:        "test.txt" + s3_constants.VersionsFolder,
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+					Extended: map[string][]byte{
+						s3_constants.ExtLatestVersionIdKey:    []byte("v1-test"),
+						s3_constants.ExtLatestVersionSizeKey:  []byte("100"),
+						s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+						s3_constants.ExtLatestVersionETagKey:  []byte("\"test-etag\""),
+					},
+				},
+			},
+		},
+	}
+
+	s3a := &S3ApiServer{
+		option: &S3ApiServerOption{
+			BucketsPath: "/buckets",
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: uint16(1000)}
+	contents := []ListEntry{}
+	_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		if cursor.maxKeys <= 0 {
+			return
+		}
+		contents = append(contents, ListEntry{Key: entry.Name})
+		cursor.maxKeys--
+	})
+
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(contents), "Should return exactly 1 object (no duplicates)")
+	assert.Equal(t, "test.txt", contents[0].Key, "Should return correct key")
+}
+
+// TestVersionedObjectsWithDeleteMarker tests that objects with delete markers are not listed
+func TestVersionedObjectsWithDeleteMarker(t *testing.T) {
+	now := time.Now().Unix()
+
+	filerClient := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				// Active versioned object
+				{
+					Name:        "active.txt" + s3_constants.VersionsFolder,
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+					Extended: map[string][]byte{
+						s3_constants.ExtLatestVersionIdKey:    []byte("v1-active"),
+						s3_constants.ExtLatestVersionSizeKey:  []byte("100"),
+						s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+						s3_constants.ExtLatestVersionETagKey:  []byte("\"etag-active\""),
+					},
+				},
+				// Deleted object (has delete marker)
+				{
+					Name:        "deleted.txt" + s3_constants.VersionsFolder,
+					IsDirectory: true,
+					Attributes: &filer_pb.FuseAttributes{
+						Mtime: now,
+					},
+					Extended: map[string][]byte{
+						s3_constants.ExtLatestVersionIdKey:          []byte("v1-deleted"),
+						s3_constants.ExtLatestVersionIsDeleteMarker: []byte("true"),
+					},
+				},
+			},
+		},
+	}
+
+	s3a := &S3ApiServer{
+		option: &S3ApiServerOption{
+			BucketsPath: "/buckets",
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: uint16(1000)}
+	contents := []ListEntry{}
+	_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		if cursor.maxKeys <= 0 {
+			return
+		}
+		contents = append(contents, ListEntry{Key: entry.Name})
+		cursor.maxKeys--
+	})
+
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(contents), "Should only return active object, not deleted")
+	assert.Equal(t, "active.txt", contents[0].Key, "Should return the active object")
+}
+
+// TestVersionedObjectsMaxKeys tests pagination with versioned objects
+func TestVersionedObjectsMaxKeys(t *testing.T) {
+	now := time.Now().Unix()
+
+	// Create 5 versioned objects
+	entries := make([]*filer_pb.Entry, 5)
+	for i := 0; i < 5; i++ {
+		entries[i] = &filer_pb.Entry{
+			Name:        fmt.Sprintf("file%d.txt"+s3_constants.VersionsFolder, i),
+			IsDirectory: true,
+			Attributes: &filer_pb.FuseAttributes{
+				Mtime: now,
+			},
+			Extended: map[string][]byte{
+				s3_constants.ExtLatestVersionIdKey:    []byte(fmt.Sprintf("v%d", i)),
+				s3_constants.ExtLatestVersionSizeKey:  []byte("100"),
+				s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+				s3_constants.ExtLatestVersionETagKey:  []byte(fmt.Sprintf("\"etag-%d\"", i)),
+			},
+		}
+	}
+
+	filerClient := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": entries,
+		},
+	}
+
+	s3a := &S3ApiServer{
+		option: &S3ApiServerOption{
+			BucketsPath: "/buckets",
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: uint16(3)}
+	contents := []ListEntry{}
+	_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		if cursor.maxKeys <= 0 {
+			return
+		}
+		contents = append(contents, ListEntry{Key: entry.Name})
+		cursor.maxKeys--
+	})
+
+	assert.NoError(t, err)
+	assert.Equal(t, 3, len(contents), "Should respect maxKeys limit")
+	assert.True(t, cursor.isTruncated, "Should set IsTruncated when there are more results")
+
+	// Verify truncation is properly set when maxKeys is exceeded
+	// (The test mock doesn't implement marker-based pagination, but we can verify
+	// that the cursor state is correct for actual pagination to work)
+	assert.True(t, cursor.isTruncated, "IsTruncated should be true when maxKeys is exhausted with more entries available")
+}
+
+// TestVersionsDirectoryNotTraversed ensures .versions directories are never traversed
+func TestVersionsDirectoryNotTraversed(t *testing.T) {
+	now := time.Now().Unix()
+	traversedDirs := make(map[string]bool)
+
+	// Custom filer client that tracks which directories are accessed
+	customClient := &customTestFilerClient{
+		testFilerClient: testFilerClient{
+			entriesByDir: map[string][]*filer_pb.Entry{
+				"/buckets/test-bucket": {
+					{
+						Name:        "object.txt" + s3_constants.VersionsFolder,
+						IsDirectory: true,
+						Attributes: &filer_pb.FuseAttributes{
+							Mtime: now,
+						},
+						Extended: map[string][]byte{
+							s3_constants.ExtLatestVersionIdKey:    []byte("v1"),
+							s3_constants.ExtLatestVersionSizeKey:  []byte("100"),
+							s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)),
+							s3_constants.ExtLatestVersionETagKey:  []byte("\"etag\""),
+						},
+					},
+				},
+				// This directory should NEVER be accessed
+				"/buckets/test-bucket/object.txt.versions": {
+					{
+						Name:        "should-not-see-this",
+						IsDirectory: false,
+					},
+				},
+			},
+		},
+		traversedDirs: &traversedDirs,
+	}
+
+	s3a := &S3ApiServer{
+		option: &S3ApiServerOption{
+			BucketsPath: "/buckets",
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: uint16(1000)}
+	contents := []ListEntry{}
+	_, err := s3a.doListFilerEntries(customClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		if cursor.maxKeys <= 0 {
+			return
+		}
+		contents = append(contents, ListEntry{Key: entry.Name})
+		cursor.maxKeys--
+	})
+
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(contents))
+
+	// Verify .versions directory was NEVER traversed
+	_, wasTraversed := traversedDirs["/buckets/test-bucket/object.txt.versions"]
+	assert.False(t, wasTraversed, ".versions directory should never be traversed")
+}
+
+// customTestFilerClient tracks which directories are accessed
+type customTestFilerClient struct {
+	testFilerClient
+	traversedDirs *map[string]bool
+}
+
+func (c *customTestFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	(*c.traversedDirs)[in.Directory] = true
+	return c.testFilerClient.ListEntries(ctx, in, opts...)
+}
@@ -76,7 +76,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 	// http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html

 	bucket, object := s3_constants.GetBucketAndObject(r)
-	glog.V(3).Infof("PutObjectHandler %s %s", bucket, object)
+	glog.V(2).Infof("PutObjectHandler bucket=%s object=%s size=%d", bucket, object, r.ContentLength)

 	_, err := validateContentMd5(r.Header)
 	if err != nil {
@@ -201,11 +201,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)

 			for _, v := range newVolumes {
-				glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
+				glog.V(1).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
 				message.NewVids = append(message.NewVids, uint32(v.Id))
 			}
 			for _, v := range deletedVolumes {
-				glog.V(0).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
+				glog.V(1).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
 				message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
 			}
 		}
@@ -197,7 +197,7 @@ func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind Ne
 	l.SetVolume(vid, v)

 	size, _, _ := v.FileStat()
-	glog.V(0).Infof("data file %s, replication=%s v=%d size=%d ttl=%s disk_id=%d",
+	glog.V(2).Infof("data file %s, replication=%s v=%d size=%d ttl=%s disk_id=%d",
 		l.Directory+"/"+volumeName+".dat", v.ReplicaPlacement, v.Version(), size, v.Ttl.String(), diskId)
 	return true
 }
@@ -257,10 +257,10 @@ func (l *DiskLocation) loadExistingVolumesWithId(needleMapKind NeedleMapKind, ld
 		}
 	}
 	l.concurrentLoadingVolumes(needleMapKind, workerNum, ldbTimeout, diskId)
-	glog.V(0).Infof("Store started on dir: %s with %d volumes max %d (disk ID: %d)", l.Directory, len(l.volumes), l.MaxVolumeCount, diskId)
+	glog.V(2).Infof("Store started on dir: %s with %d volumes max %d (disk ID: %d)", l.Directory, len(l.volumes), l.MaxVolumeCount, diskId)

 	l.loadAllEcShards()
-	glog.V(0).Infof("Store started on dir: %s with %d ec shards (disk ID: %d)", l.Directory, len(l.ecVolumes), diskId)
+	glog.V(2).Infof("Store started on dir: %s with %d ec shards (disk ID: %d)", l.Directory, len(l.ecVolumes), diskId)

 }

@@ -108,7 +108,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		if err == nil {
 			v.volumeInfo.Version = uint32(v.SuperBlock.Version)
 		}
-		glog.V(0).Infof("readSuperBlock volume %d version %v", v.Id, v.SuperBlock.Version)
+		glog.V(2).Infof("readSuperBlock volume %d version %v", v.Id, v.SuperBlock.Version)
 		if v.HasRemoteFile() {
 			// maybe temporary network problem
 			glog.Errorf("readSuperBlock remote volume %d: %v", v.Id, err)
@@ -149,7 +149,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		// storage tier, and download to local storage, which may cause the
 		// capactiy overloading.
 		if !v.HasRemoteFile() {
-			glog.V(0).Infof("checking volume data integrity for volume %d", v.Id)
+			glog.V(2).Infof("checking volume data integrity for volume %d", v.Id)
 			if v.lastAppendAtNs, err = CheckVolumeDataIntegrity(v, indexFile); err != nil {
 				v.noWriteOrDelete = true
 				glog.V(0).Infof("volumeDataIntegrityChecking failed %v", err)
@@ -164,10 +164,10 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 			switch needleMapKind {
 			case NeedleMapInMemory:
 				if v.tmpNm != nil {
-					glog.V(0).Infof("updating memory compact index %s ", v.FileName(".idx"))
+					glog.V(2).Infof("updating memory compact index %s ", v.FileName(".idx"))
 					err = v.tmpNm.UpdateNeedleMap(v, indexFile, nil, 0)
 				} else {
-					glog.V(0).Infoln("loading memory index", v.FileName(".idx"), "to memory")
+					glog.V(2).Infoln("loading memory index", v.FileName(".idx"), "to memory")
 					if v.nm, err = LoadCompactNeedleMap(indexFile); err != nil {
 						glog.V(0).Infof("loading index %s to memory error: %v", v.FileName(".idx"), err)
 					}
@@ -414,7 +414,7 @@ func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool {
 			return false
 		}
 	}
-	glog.V(0).Infoln("Volume", vid, "becomes writable")
+	glog.V(1).Infoln("Volume", vid, "becomes writable")
 	vl.writables = append(vl.writables, vid)
 	return true
 }
@@ -74,6 +74,8 @@ type grpcState struct {
 	lastWorkerInfo  *types.WorkerData
 	reconnectStop   chan struct{}
 	streamExit      chan struct{}
+	streamFailed    chan struct{} // Signals when stream has failed
+	regWait         chan *worker_pb.RegistrationResponse
 }

 // NewGrpcAdminClient creates a new gRPC admin client
@@ -98,6 +100,36 @@ func NewGrpcAdminClient(adminAddress string, workerID string, dialOption grpc.Di
 	return c
 }

+// drainAndCloseRegWaitChannel drains any pending messages from the regWait channel
+// and then safely closes it. This prevents losing RegistrationResponse messages
+// that were sent before the channel is closed.
+func drainAndCloseRegWaitChannel(ch *chan *worker_pb.RegistrationResponse) {
+	if ch == nil || *ch == nil {
+		return
+	}
+	for {
+		select {
+		case <-*ch:
+			// continue draining until channel is empty
+		default:
+			close(*ch)
+			*ch = nil
+			return
+		}
+	}
+}
+
+// safeCloseChannel safely closes a channel and sets it to nil to prevent double-close panics.
+// NOTE: This function is NOT thread-safe. It is safe to use in this codebase because all calls
+// are serialized within the managerLoop goroutine. If this function is used in concurrent contexts
+// in the future, synchronization (e.g., sync.Mutex) should be added.
+func (c *GrpcAdminClient) safeCloseChannel(chPtr *chan struct{}) {
+	if *chPtr != nil {
+		close(*chPtr)
+		*chPtr = nil
+	}
+}
+
 func (c *GrpcAdminClient) managerLoop() {
 	state := &grpcState{shouldReconnect: true}

@@ -129,7 +161,14 @@ out:
 				req.Resp <- nil
 				continue
 			}
-			err := c.sendRegistration(req.Worker)
+			// Capture channel pointers to avoid race condition with reconnect
+			streamFailedCh := state.streamFailed
+			regWaitCh := state.regWait
+			if streamFailedCh == nil || regWaitCh == nil {
+				req.Resp <- fmt.Errorf("stream not ready for registration")
+				continue
+			}
+			err := c.sendRegistration(req.Worker, streamFailedCh, regWaitCh)
 			req.Resp <- err
 		case ActionQueryConnected:
 			respCh := cmd.data.(chan bool)
@@ -214,14 +253,18 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error {
 	// Start stream handlers BEFORE sending registration
 	// This ensures handleIncoming is ready to receive the registration response
 	s.streamExit = make(chan struct{})
+	s.streamFailed = make(chan struct{})
+	s.regWait = make(chan *worker_pb.RegistrationResponse, 1)
 	go handleOutgoing(s.stream, s.streamExit, c.outgoing, c.cmds)
-	go handleIncoming(c.workerID, s.stream, s.streamExit, c.incoming, c.cmds)
+	go handleIncoming(c.workerID, s.stream, s.streamExit, c.incoming, c.cmds, s.streamFailed, s.regWait)

 	// Always check for worker info and send registration immediately as the very first message
 	if s.lastWorkerInfo != nil {
 		// Send registration via the normal outgoing channel and wait for response via incoming
-		if err := c.sendRegistration(s.lastWorkerInfo); err != nil {
-			close(s.streamExit)
+		if err := c.sendRegistration(s.lastWorkerInfo, s.streamFailed, s.regWait); err != nil {
+			c.safeCloseChannel(&s.streamExit)
+			c.safeCloseChannel(&s.streamFailed)
+			drainAndCloseRegWaitChannel(&s.regWait)
 			s.streamCancel()
 			s.conn.Close()
 			s.connected = false
@@ -240,9 +283,9 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error {
 // reconnect attempts to re-establish the connection
 func (c *GrpcAdminClient) reconnect(s *grpcState) error {
 	// Clean up existing connection completely
-	if s.streamExit != nil {
-		close(s.streamExit)
-	}
+	c.safeCloseChannel(&s.streamExit)
+	c.safeCloseChannel(&s.streamFailed)
+	drainAndCloseRegWaitChannel(&s.regWait)
 	if s.streamCancel != nil {
 		s.streamCancel()
 	}
@@ -315,32 +358,70 @@ func handleOutgoing(
 	streamExit <-chan struct{},
 	outgoing <-chan *worker_pb.WorkerMessage,
 	cmds chan<- grpcCommand) {
-
-	msgCh := make(chan *worker_pb.WorkerMessage)
+	msgCh := make(chan *worker_pb.WorkerMessage, 1)
 	errCh := make(chan error, 1) // Buffered to prevent blocking if the manager is busy
-	// Goroutine to handle blocking stream.Recv() and simultaneously handle exit
-	// signals
+
+	// Goroutine that reads from msgCh and performs the blocking stream.Send() calls.
 	go func() {
 		for msg := range msgCh {
 			if err := stream.Send(msg); err != nil {
 				errCh <- err
-				return // Exit the receiver goroutine on error/EOF
+				return
 			}
 		}
 		close(errCh)
 	}()

-	for msg := range outgoing {
-		select {
-		case msgCh <- msg:
-		case err := <-errCh:
+	// Helper function to handle stream errors and cleanup
+	handleStreamError := func(err error) {
+		if err != nil {
 			glog.Errorf("Failed to send message to admin: %v", err)
-			cmds <- grpcCommand{action: ActionStreamError, data: err}
-			return
+			select {
+			case cmds <- grpcCommand{action: ActionStreamError, data: err}:
+				// Successfully queued
+			default:
+				// Manager busy, queue asynchronously to avoid blocking
+				glog.V(2).Infof("Manager busy, queuing stream error asynchronously from outgoing handler: %v", err)
+				go func(e error) {
+					select {
+					case cmds <- grpcCommand{action: ActionStreamError, data: e}:
+					case <-time.After(2 * time.Second):
+						glog.Warningf("Failed to send stream error to manager from outgoing handler, channel blocked: %v", e)
+					}
+				}(err)
+			}
+		}
+	}
+
+	// Helper function to cleanup resources
+	cleanup := func() {
+		close(msgCh)
+		<-errCh
+	}
+
+	for {
+		select {
 		case <-streamExit:
-			close(msgCh)
-			<-errCh
+			cleanup()
 			return
+		case err := <-errCh:
+			handleStreamError(err)
+			return
+		case msg, ok := <-outgoing:
+			if !ok {
+				cleanup()
+				return
+			}
+			select {
+			case msgCh <- msg:
+				// Message queued successfully
+			case <-streamExit:
+				cleanup()
+				return
+			case err := <-errCh:
+				handleStreamError(err)
+				return
+			}
 		}
 	}
 }
@@ -351,10 +432,15 @@ func handleIncoming(
 	stream worker_pb.WorkerService_WorkerStreamClient,
 	streamExit <-chan struct{},
 	incoming chan<- *worker_pb.AdminMessage,
-	cmds chan<- grpcCommand) {
+	cmds chan<- grpcCommand,
+	streamFailed chan<- struct{},
+	regWait chan<- *worker_pb.RegistrationResponse) {
 	glog.V(1).Infof("INCOMING HANDLER STARTED: Worker %s incoming message handler started", workerID)
 	msgCh := make(chan *worker_pb.AdminMessage)
 	errCh := make(chan error, 1) // Buffered to prevent blocking if the manager is busy
+	// regWait is buffered with size 1 so that the registration response can be sent
+	// even if the receiver goroutine has not yet started waiting on the channel.
+	// This non-blocking send pattern avoids a race between sendRegistration and handleIncoming.
 	// Goroutine to handle blocking stream.Recv() and simultaneously handle exit
 	// signals
 	go func() {
@@ -376,7 +462,19 @@ func handleIncoming(
 			// Message successfully received from the stream
 			glog.V(4).Infof("MESSAGE RECEIVED: Worker %s received message from admin server: %T", workerID, msg.Message)

-			// Route message to waiting goroutines or general handler (original select logic)
+			// If this is a registration response, also publish to the registration waiter.
+			// regWait is buffered (size 1) so that the response can be sent even if sendRegistration
+			// hasn't started waiting yet, preventing a race condition between the two goroutines.
+			if rr := msg.GetRegistrationResponse(); rr != nil {
+				select {
+				case regWait <- rr:
+					glog.V(3).Infof("REGISTRATION RESPONSE: Worker %s routed registration response to waiter", workerID)
+				default:
+					glog.V(2).Infof("REGISTRATION RESPONSE DROPPED: Worker %s registration response dropped (no waiter)", workerID)
+				}
+			}
+
+			// Route message to general handler.
 			select {
 			case incoming <- msg:
 				glog.V(3).Infof("MESSAGE ROUTED: Worker %s successfully routed message to handler", workerID)
@@ -392,8 +490,27 @@ func handleIncoming(
 				glog.Errorf("RECEIVE ERROR: Worker %s failed to receive message from admin: %v", workerID, err)
 			}

-			// Report the failure as a command to the managerLoop (blocking)
-			cmds <- grpcCommand{action: ActionStreamError, data: err}
+			// Signal that stream has failed (non-blocking)
+			select {
+			case streamFailed <- struct{}{}:
+			default:
+			}
+
+			// Report the failure as a command to the managerLoop.
+			// Try non-blocking first; if the manager is busy and the channel is full,
+			// fall back to an asynchronous blocking send so the error is not lost.
+			select {
+			case cmds <- grpcCommand{action: ActionStreamError, data: err}:
+			default:
+				glog.V(2).Infof("Manager busy, queuing stream error asynchronously: %v", err)
+				go func(e error) {
+					select {
+					case cmds <- grpcCommand{action: ActionStreamError, data: e}:
+					case <-time.After(2 * time.Second):
+						glog.Warningf("Failed to send stream error to manager, channel blocked: %v", e)
+					}
+				}(err)
+			}

 			// Exit the main handler loop
 			glog.V(1).Infof("INCOMING HANDLER STOPPED: Worker %s stopping incoming handler due to stream error", workerID)
@@ -425,7 +542,7 @@ func (c *GrpcAdminClient) handleDisconnect(cmd grpcCommand, s *grpcState) {
 	}

 	// Send shutdown signal to stop reconnection loop
-	close(s.reconnectStop)
+	c.safeCloseChannel(&s.reconnectStop)

 	s.connected = false
 	s.shouldReconnect = false
@@ -450,7 +567,9 @@ func (c *GrpcAdminClient) handleDisconnect(cmd grpcCommand, s *grpcState) {
 	}

 	// Send shutdown signal to stop handlers loop
-	close(s.streamExit)
+	c.safeCloseChannel(&s.streamExit)
+	c.safeCloseChannel(&s.streamFailed)
+	drainAndCloseRegWaitChannel(&s.regWait)

 	// Cancel stream context
 	if s.streamCancel != nil {
@@ -486,7 +605,7 @@ func (c *GrpcAdminClient) RegisterWorker(worker *types.WorkerData) error {
 }

 // sendRegistration sends the registration message and waits for response
-func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData) error {
+func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData, streamFailed <-chan struct{}, regWait <-chan *worker_pb.RegistrationResponse) error {
 	capabilities := make([]string, len(worker.Capabilities))
 	for i, cap := range worker.Capabilities {
 		capabilities[i] = string(cap)
@@ -510,24 +629,30 @@ func (c *GrpcAdminClient) sendRegistration(worker *types.WorkerData) error {
 	case c.outgoing <- msg:
 	case <-time.After(5 * time.Second):
 		return fmt.Errorf("failed to send registration message: timeout")
+	case <-streamFailed:
+		return fmt.Errorf("stream failed before registration message could be sent")
 	}

 	// Wait for registration response
-	timeout := time.NewTimer(10 * time.Second)
+	// Use longer timeout for reconnections since admin server might be busy
+	timeout := time.NewTimer(30 * time.Second)
 	defer timeout.Stop()

 	for {
 		select {
-		case response := <-c.incoming:
-			if regResp := response.GetRegistrationResponse(); regResp != nil {
-				if regResp.Success {
-					glog.Infof("Worker registered successfully: %s", regResp.Message)
-					return nil
-				}
-				return fmt.Errorf("registration failed: %s", regResp.Message)
+		case regResp, ok := <-regWait:
+			if !ok || regResp == nil {
+				return fmt.Errorf("registration failed: channel closed unexpectedly")
 			}
+			if regResp.Success {
+				glog.Infof("Worker registered successfully: %s", regResp.Message)
+				return nil
+			}
+			return fmt.Errorf("registration failed: %s", regResp.Message)
+		case <-streamFailed:
+			return fmt.Errorf("registration failed: stream closed by server")
 		case <-timeout.C:
-			return fmt.Errorf("registration timeout")
+			return fmt.Errorf("registration failed: timeout waiting for response")
 		}
 	}
 }