add ifmatch support (#885)

Co-authored-by: loverustfs <hello@rustfs.com>

.docker/observability/docker-compose.yml

@@ -16,7 +16,7 @@ services:
 
   tempo-init:
     image: busybox:latest
-    command: ["sh", "-c", "chown -R 10001:10001 /var/tempo"]
+    command: [ "sh", "-c", "chown -R 10001:10001 /var/tempo" ]
     volumes:
       - ./tempo-data:/var/tempo
     user: root
@@ -39,7 +39,7 @@ services:
       - otel-network
 
   otel-collector:
-    image: otel/opentelemetry-collector-contrib:0.129.1
+    image: otel/opentelemetry-collector-contrib:latest
     environment:
       - TZ=Asia/Shanghai
     volumes:
@@ -55,7 +55,7 @@ services:
     networks:
       - otel-network
   jaeger:
-    image: jaegertracing/jaeger:2.8.0
+    image: jaegertracing/jaeger:latest
     environment:
       - TZ=Asia/Shanghai
     ports:
@@ -65,17 +65,21 @@ services:
     networks:
       - otel-network
   prometheus:
-    image: prom/prometheus:v3.4.2
+    image: prom/prometheus:latest
     environment:
       - TZ=Asia/Shanghai
     volumes:
       - ./prometheus.yml:/etc/prometheus/prometheus.yml
     ports:
       - "9090:9090"
+    command:
+      - '--config.file=/etc/prometheus/prometheus.yml'
+      - '--web.enable-otlp-receiver' # Enable OTLP
+      - '--enable-feature=promql-experimental-functions' # Enable info()
     networks:
       - otel-network
   loki:
-    image: grafana/loki:3.5.1
+    image: grafana/loki:latest
     environment:
       - TZ=Asia/Shanghai
     volumes:
@@ -86,7 +90,7 @@ services:
     networks:
       - otel-network
   grafana:
-    image: grafana/grafana:12.0.2
+    image: grafana/grafana:latest
     ports:
       - "3000:3000"  # Web UI
     volumes:

.docker/observability/grafana-datasources.yaml

@@ -29,4 +29,80 @@ datasources:
       serviceMap:
         datasourceUid: prometheus
       streamingEnabled:
-        search: true
+        search: true
+      tracesToLogsV2:
+        # Field with an internal link pointing to a logs data source in Grafana.
+        # datasourceUid value must match the uid value of the logs data source.
+        datasourceUid: 'loki'
+        spanStartTimeShift: '-1h'
+        spanEndTimeShift: '1h'
+        tags: [ 'job', 'instance', 'pod', 'namespace' ]
+        filterByTraceID: false
+        filterBySpanID: false
+        customQuery: true
+        query: 'method="$${__span.tags.method}"'
+        tracesToMetrics:
+          datasourceUid: 'prom'
+          spanStartTimeShift: '-1h'
+          spanEndTimeShift: '1h'
+          tags: [ { key: 'service.name', value: 'service' }, { key: 'job' } ]
+          queries:
+            - name: 'Sample query'
+              query: 'sum(rate(traces_spanmetrics_latency_bucket{$$__tags}[5m]))'
+        tracesToProfiles:
+          datasourceUid: 'grafana-pyroscope-datasource'
+          tags: [ 'job', 'instance', 'pod', 'namespace' ]
+          profileTypeId: 'process_cpu:cpu:nanoseconds:cpu:nanoseconds'
+          customQuery: true
+          query: 'method="$${__span.tags.method}"'
+        serviceMap:
+          datasourceUid: 'prometheus'
+        nodeGraph:
+          enabled: true
+        search:
+          hide: false
+        traceQuery:
+          timeShiftEnabled: true
+          spanStartTimeShift: '-1h'
+          spanEndTimeShift: '1h'
+        spanBar:
+          type: 'Tag'
+          tag: 'http.path'
+        streamingEnabled:
+          search: true
+  - name: Jaeger
+    type: jaeger
+    uid: Jaeger
+    url: http://jaeger:16686
+    basicAuth: false
+    access: proxy
+    readOnly: false
+    isDefault: false
+    jsonData:
+      tracesToLogsV2:
+        # Field with an internal link pointing to a logs data source in Grafana.
+        # datasourceUid value must match the uid value of the logs data source.
+        datasourceUid: 'loki'
+        spanStartTimeShift: '1h'
+        spanEndTimeShift: '-1h'
+        tags: [ 'job', 'instance', 'pod', 'namespace' ]
+        filterByTraceID: false
+        filterBySpanID: false
+        customQuery: true
+        query: 'method="$${__span.tags.method}"'
+      tracesToMetrics:
+        datasourceUid: 'prom'
+        spanStartTimeShift: '1h'
+        spanEndTimeShift: '-1h'
+        tags: [ { key: 'service.name', value: 'service' }, { key: 'job' } ]
+        queries:
+          - name: 'Sample query'
+            query: 'sum(rate(traces_spanmetrics_latency_bucket{$$__tags}[5m]))'
+      nodeGraph:
+        enabled: true
+      traceQuery:
+        timeShiftEnabled: true
+        spanStartTimeShift: '1h'
+        spanEndTimeShift: '-1h'
+      spanBar:
+        type: 'None'

.docker/observability/loki-config.yaml

@@ -63,6 +63,7 @@ ruler:
 frontend:
   encoding: protobuf
 
+
 # By default, Loki will send anonymous, but uniquely-identifiable usage and configuration
 # analytics to Grafana Labs. These statistics are sent to https://stats.grafana.org/
 #

.docker/observability/otel-collector-config.yaml

@@ -43,7 +43,6 @@ exporters:
     send_timestamps: true  # 发送时间戳
     # enable_open_metrics: true
   otlphttp/loki: # Loki 导出器，用于日志数据
-    # endpoint: "http://loki:3100/otlp/v1/logs"
     endpoint: "http://loki:3100/otlp/v1/logs"
     tls:
       insecure: true

.docker/observability/prometheus.yml

@@ -13,16 +13,43 @@
 # limitations under the License.
 
 global:
-  scrape_interval: 5s  # 刮取间隔
+  scrape_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
 
 scrape_configs:
   - job_name: 'otel-collector'
     static_configs:
-      - targets: [ 'otel-collector:8888' ]  # 从 Collector 刮取指标
+      - targets: [ 'otel-collector:8888' ]  # Scrape metrics from Collector
   - job_name: 'otel-metrics'
     static_configs:
-      - targets: [ 'otel-collector:8889' ]  # 应用指标
+      - targets: [ 'otel-collector:8889' ]  # Application indicators
   - job_name: 'tempo'
     static_configs:
-      - targets: [ 'tempo:3200' ]
-      
+      - targets: [ 'tempo:3200' ]  # Scrape metrics from Tempo
+
+otlp:
+  # Recommended attributes to be promoted to labels.
+  promote_resource_attributes:
+    - service.instance.id
+    - service.name
+    - service.namespace
+    - cloud.availability_zone
+    - cloud.region
+    - container.name
+    - deployment.environment.name
+    - k8s.cluster.name
+    - k8s.container.name
+    - k8s.cronjob.name
+    - k8s.daemonset.name
+    - k8s.deployment.name
+    - k8s.job.name
+    - k8s.namespace.name
+    - k8s.pod.name
+    - k8s.replicaset.name
+    - k8s.statefulset.name
+  # Ingest OTLP data keeping all characters in metric/label names.
+  translation_strategy: NoUTF8EscapingWithSuffixes
+
+storage:
+  # OTLP is a push-based protocol, Out of order samples is a common scenario.
+  tsdb:
+    out_of_order_time_window: 30m

.github/actions/setup/action.yml

@@ -52,24 +52,19 @@ runs:
         sudo apt-get install -y \
           musl-tools \
           build-essential \
-          lld \
-          libdbus-1-dev \
-          libwayland-dev \
-          libwebkit2gtk-4.1-dev \
-          libxdo-dev \
           pkg-config \
           libssl-dev
 
     - name: Install protoc
       uses: arduino/setup-protoc@v3
       with:
-        version: "31.1"
+        version: "33.1"
         repo-token: ${{ inputs.github-token }}
 
     - name: Install flatc
       uses: Nugine/setup-flatc@v1
       with:
-        version: "25.2.10"
+        version: "25.9.23"
 
     - name: Install Rust toolchain
       uses: dtolnay/rust-toolchain@stable

.github/dependabot.yml

@@ -22,8 +22,18 @@ updates:
   - package-ecosystem: "cargo" # See documentation for possible values
     directory: "/" # Location of package manifests
     schedule:
-      interval: "monthly"
+      interval: "weekly"
+      day: "monday"
+      timezone: "Asia/Shanghai"
+      time: "08:00"
     groups:
+      s3s:
+        update-types:
+          - "minor"
+          - "patch"
+        patterns:
+          - "s3s"
+          - "s3s-*"
       dependencies:
         patterns:
           - "*"

Cargo.toml

@@ -97,15 +97,15 @@ async-channel = "2.5.0"
 async-compression = { version = "0.4.19" }
 async-recursion = "1.1.1"
 async-trait = "0.1.89"
-axum = "0.8.6"
-axum-extra = "0.12.1"
-axum-server = { version = "0.7.2", features = ["tls-rustls-no-provider"], default-features = false }
+axum = "0.8.7"
+axum-extra = "0.12.2"
+axum-server = { version = "0.7.3", features = ["tls-rustls-no-provider"], default-features = false }
 futures = "0.3.31"
 futures-core = "0.3.31"
 futures-util = "0.3.31"
-hyper = { version = "1.7.0", features = ["http2", "http1", "server"] }
+hyper = { version = "1.8.1", features = ["http2", "http1", "server"] }
 hyper-rustls = { version = "0.27.7", default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "ring", "webpki-roots"] }
-hyper-util = { version = "0.1.17", features = ["tokio", "server-auto", "server-graceful"] }
+hyper-util = { version = "0.1.18", features = ["tokio", "server-auto", "server-graceful"] }
 http = "1.3.1"
 http-body = "1.0.1"
 reqwest = { version = "0.12.24", default-features = false, features = ["rustls-tls-webpki-roots", "charset", "http2", "system-proxy", "stream", "json", "blocking"] }
@@ -122,39 +122,36 @@ tower = { version = "0.5.2", features = ["timeout"] }
 tower-http = { version = "0.6.6", features = ["cors"] }
 
 # Serialization and Data Formats
-bytes = { version = "1.10.1", features = ["serde"] }
-bytesize = "2.1.0"
+bytes = { version = "1.11.0", features = ["serde"] }
+bytesize = "2.2.0"
 byteorder = "1.5.0"
 flatbuffers = "25.9.23"
 form_urlencoded = "1.2.2"
 prost = "0.14.1"
-quick-xml = "0.38.3"
-rmcp = { version = "0.8.4" }
+quick-xml = "0.38.4"
+rmcp = { version = "0.8.5" }
 rmp = { version = "0.8.14" }
 rmp-serde = { version = "1.3.0" }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = { version = "1.0.145", features = ["raw_value"] }
 serde_urlencoded = "0.7.1"
-schemars = "1.0.5"
+schemars = "1.1.0"
 
 # Cryptography and Security
-aes-gcm = { version = "0.10.3", features = ["std"] }
-argon2 = { version = "0.5.3", features = ["std"] }
+aes-gcm = { version = "0.11.0-rc.2", features = ["rand_core"] }
+argon2 = { version = "0.6.0-rc.2", features = ["std"] }
 blake3 = { version = "1.8.2" }
-chacha20poly1305 = { version = "0.10.1" }
-crc-fast = "1.3.0"
-crc32c = "0.6.8"
-crc32fast = "1.5.0"
-crc64fast-nvme = "1.2.0"
-hmac = "0.12.1"
-jsonwebtoken = { version = "10.1.0", features = ["rust_crypto"] }
-pbkdf2 = "0.12.2"
-rsa = { version = "0.9.8" }
+chacha20poly1305 = { version = "0.11.0-rc.2" }
+crc-fast = "1.6.0"
+hmac = { version = "0.13.0-rc.3" }
+jsonwebtoken = { version = "10.2.0", features = ["rust_crypto"] }
+pbkdf2 = "0.13.0-rc.2"
+rsa = { version = "0.10.0-rc.10" }
 rustls = { version = "0.23.35", features = ["ring", "logging", "std", "tls12"], default-features = false }
 rustls-pemfile = "2.2.0"
 rustls-pki-types = "1.13.0"
-sha1 = "0.10.6"
-sha2 = "0.10.9"
+sha1 = "0.11.0-rc.3"
+sha2 = "0.11.0-rc.3"
 zeroize = { version = "1.8.2", features = ["derive"] }
 
 # Time and Date
@@ -169,8 +166,8 @@ astral-tokio-tar = "0.5.6"
 atoi = "2.0.0"
 atomic_enum = "0.3.0"
 aws-config = { version = "1.8.10" }
-aws-credential-types = { version = "1.2.8" }
-aws-sdk-s3 = { version = "1.110.0", default-features = false, features = ["sigv4a", "rustls", "rt-tokio"] }
+aws-credential-types = { version = "1.2.9" }
+aws-sdk-s3 = { version = "1.112.0", default-features = false, features = ["sigv4a", "rustls", "rt-tokio"] }
 aws-smithy-types = { version = "1.3.4" }
 base64 = "0.22.1"
 base64-simd = "0.8.0"
@@ -178,17 +175,18 @@ brotli = "8.0.2"
 cfg-if = "1.0.4"
 clap = { version = "4.5.51", features = ["derive", "env"] }
 const-str = { version = "0.7.0", features = ["std", "proc"] }
-convert_case = "0.8.0"
+convert_case = "0.9.0"
 criterion = { version = "0.7", features = ["html_reports"] }
 crossbeam-queue = "0.3.12"
 datafusion = "50.3.0"
 derive_builder = "0.20.2"
 enumset = "1.1.10"
+faster-hex = "0.10.0"
 flate2 = "1.1.5"
-flexi_logger = { version = "0.31.7", features = ["trc", "dont_minimize_extra_stacks", "compress", "kv"] }
+flexi_logger = { version = "0.31.7", features = ["trc", "dont_minimize_extra_stacks", "compress", "kv", "json"] }
 glob = "0.3.3"
-google-cloud-storage = "1.2.0"
-google-cloud-auth = "1.1.0"
+google-cloud-storage = "1.4.0"
+google-cloud-auth = "1.2.0"
 hashbrown = { version = "0.16.0", features = ["serde", "rayon"] }
 heed = { version = "0.22.0" }
 hex-simd = "0.8.0"
@@ -196,14 +194,13 @@ highway = { version = "1.3.0" }
 ipnetwork = { version = "0.21.1", features = ["serde"] }
 lazy_static = "1.5.0"
 libc = "0.2.177"
-libsystemd = { version = "0.7.2" }
+libsystemd = "0.7.2"
 local-ip-address = "0.6.5"
 lz4 = "1.28.1"
 matchit = "0.9.0"
-md-5 = "0.10.6"
+md-5 = "0.11.0-rc.3"
 md5 = "0.8.0"
 metrics = "0.24.2"
-metrics-exporter-opentelemetry = "0.1.2"
 mime_guess = "2.0.5"
 moka = { version = "0.12.11", features = ["future"] }
 netif = "0.1.6"
@@ -218,15 +215,14 @@ path-absolutize = "3.1.1"
 path-clean = "1.0.1"
 pin-project-lite = "0.2.16"
 pretty_assertions = "1.4.1"
-rand = "0.9.2"
+rand = { version = "0.10.0-rc.5", features = ["serde"] }
 rayon = "1.11.0"
 reed-solomon-simd = { version = "3.1.0" }
 regex = { version = "1.12.2" }
 rumqttc = { version = "0.25.0" }
 rust-embed = { version = "8.9.0" }
 rustc-hash = { version = "2.1.1" }
-s3s = { version = "0.12.0-rc.3", features = ["minio"] }
-scopeguard = "1.2.0"
+s3s = { git = "https://github.com/s3s-project/s3s.git", rev = "ba9f902", version = "0.12.0-rc.3", features = ["minio"] }
 serial_test = "3.2.0"
 shadow-rs = { version = "1.4.0", default-features = false }
 siphasher = "1.0.1"
@@ -253,7 +249,7 @@ urlencoding = "2.1.3"
 uuid = { version = "1.18.1", features = ["v4", "fast-rng", "macro-diagnostics"] }
 vaultrs = { version = "0.7.4" }
 walkdir = "2.5.0"
-wildmatch = { version = "2.5.0", features = ["serde"] }
+wildmatch = { version = "2.6.1", features = ["serde"] }
 winapi = { version = "0.3.9" }
 xxhash-rust = { version = "0.8.15", features = ["xxh64", "xxh3"] }
 zip = "6.0.0"
@@ -262,7 +258,7 @@ zstd = "0.13.3"
 # Observability and Metrics
 opentelemetry = { version = "0.31.0" }
 opentelemetry-appender-tracing = { version = "0.31.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes", "spec_unstable_logs_enabled"] }
-opentelemetry-otlp = { version = "0.31.0", default-features = false, features = ["grpc-tonic", "gzip-tonic", "trace", "metrics", "logs", "internal-logs"] }
+opentelemetry-otlp = { version = "0.31.0", features = ["http-proto", "zstd-http"] }
 opentelemetry_sdk = { version = "0.31.0" }
 opentelemetry-semantic-conventions = { version = "0.31.0", features = ["semconv_experimental"] }
 opentelemetry-stdout = { version = "0.31.0" }
@@ -280,7 +276,7 @@ mimalloc = "0.1"
 
 
 [workspace.metadata.cargo-shear]
-ignored = ["rustfs", "rustfs-mcp", "tokio-test", "scopeguard"]
+ignored = ["rustfs", "rustfs-mcp", "tokio-test"]
 
 [profile.release]
 opt-level = 3

Dockerfile

@@ -64,8 +64,12 @@ COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 COPY --from=build /build/rustfs /usr/bin/rustfs
 COPY entrypoint.sh /entrypoint.sh
 
-RUN chmod +x /usr/bin/rustfs /entrypoint.sh && \
+RUN chmod +x /usr/bin/rustfs /entrypoint.sh
+
+RUN addgroup -g 10001 -S rustfs && \
+    adduser -u 10001 -G rustfs -S rustfs -D && \
     mkdir -p /data /logs && \
+    chown -R rustfs:rustfs /data /logs && \
     chmod 0750 /data /logs
 
 ENV RUSTFS_ADDRESS=":9000" \
@@ -78,12 +82,14 @@ ENV RUSTFS_ADDRESS=":9000" \
     RUSTFS_CONSOLE_CORS_ALLOWED_ORIGINS="*" \
     RUSTFS_VOLUMES="/data" \
     RUST_LOG="warn" \
-    RUSTFS_OBS_LOG_DIRECTORY="/logs" \
-    RUSTFS_SINKS_FILE_PATH="/logs"
-
+    RUSTFS_OBS_LOG_DIRECTORY="/logs" 
+    
 EXPOSE 9000 9001
+
 VOLUME ["/data", "/logs"]
 
+USER rustfs
+
 ENTRYPOINT ["/entrypoint.sh"]
 
 CMD ["rustfs"]

Dockerfile.source

@@ -167,7 +167,6 @@ ENV RUSTFS_ADDRESS=":9000" \
     RUSTFS_VOLUMES="/data" \
     RUST_LOG="warn" \
     RUSTFS_OBS_LOG_DIRECTORY="/logs" \
-    RUSTFS_SINKS_FILE_PATH="/logs" \
     RUSTFS_USERNAME="rustfs" \
     RUSTFS_GROUPNAME="rustfs" \
     RUSTFS_UID="1000" \

README.md

@@ -1,6 +1,6 @@
 [![RustFS](https://rustfs.com/images/rustfs-github.png)](https://rustfs.com)
 
-<p align="center">RustFS is a high-performance distributed object storage software built using Rust</p>
+<p align="center">RustFS is a high-performance, distributed object storage system built in Rust.</p>
 
 <p align="center">
   <a href="https://github.com/rustfs/rustfs/actions/workflows/ci.yml"><img alt="CI" src="https://github.com/rustfs/rustfs/actions/workflows/ci.yml/badge.svg" /></a>
@@ -29,13 +29,13 @@ English | <a href="https://github.com/rustfs/rustfs/blob/main/README_ZH.md">简
   <a href="https://readme-i18n.com/rustfs/rustfs?lang=ru">Русский</a>
 </p>
 
-RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages
-worldwide. Along with MinIO, it shares a range of advantages such as simplicity, S3 compatibility, open-source nature,
+RustFS is a high-performance, distributed object storage system built in Rust., one of the most popular languages
+worldwide. RustFS combines the simplicity of MinIO with the memory safety and performance of Rust., S3 compatibility, open-source nature,
 support for data lakes, AI, and big data. Furthermore, it has a better and more user-friendly open-source license in
 comparison to other storage systems, being constructed under the Apache license. As Rust serves as its foundation,
 RustFS provides faster speed and safer distributed features for high-performance object storage.
 
-> ⚠️ **RustFS is under rapid development. Do NOT use in production environments!**
+> ⚠️ **Current Status: Beta / Technical Preview. Not yet recommended for critical production workloads.**
 
 ## Features
 
@@ -65,8 +65,8 @@ Stress test server parameters
 |---------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------|
 | Powerful Console                                                                | Simple and useless Console                                                                                              |
 | Developed based on Rust language, memory is safer                               | Developed in Go or C, with potential issues like memory GC/leaks                                                        |
-| Does not report logs to third-party countries                                   | Reporting logs to other third countries may violate national security laws                                              |
-| Licensed under Apache, more business-friendly                                   | AGPL V3 License and other License, polluted open source and License traps, infringement of intellectual property rights |
+| Guaranteed Data Sovereignty: No telemetry or unauthorized data egress       | Reporting logs to other third countries may violate national security laws                                              |
+| Permissive Apache 2.0 License     | AGPL V3 License and other License, polluted open source and License traps, infringement of intellectual property rights |
 | Comprehensive S3 support, works with domestic and international cloud providers | Full support for S3, but no local cloud vendor support                                                                  |
 | Rust-based development, strong support for secure and innovative devices        | Poor support for edge gateways and secure innovative devices                                                            |
 | Stable commercial prices, free community support                                | High pricing, with costs up to $250,000 for 1PiB                                                                        |
@@ -84,15 +84,20 @@ To get started with RustFS, follow these steps:
 
 2. **Docker Quick Start (Option 2)​​**
 
+RustFS container run as non-root user `rustfs` with id `1000`, if you run docker with `-v` to mount host directory into docker container, please make sure the owner of host directory has been changed to `1000`, otherwise you will encounter permission denied error.
+
   ```bash
    # create data and logs directories
    mkdir -p data logs
 
-   # using latest alpha version
-   docker run -d -p 9000:9000 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:alpha
+   # change the owner of those two ditectories
+   chown -R 10001:10001 data logs
 
-   # Specific version
-   docker run -d -p 9000:9000 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0.alpha.45
+   # using latest version
+   docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
+
+   # using specific version
+   docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0.alpha.68
    ```
 
 For docker installation, you can also run the container with docker compose. With the `docker-compose.yml` file under
@@ -139,6 +144,8 @@ observability. If you want to start redis as well as nginx container, you can sp
    make help-docker                      # Show all Docker-related commands
    ```
 
+   > **Heads-up (macOS cross-compilation)**: macOS keeps the default `ulimit -n` at 256, so `cargo zigbuild` or `./build-rustfs.sh --platform ...` may fail with `ProcessFdQuotaExceeded` when targeting Linux. The build script now tries to raise the limit automatically, but if you still see the warning, run `ulimit -n 4096` (or higher) in your shell before building.
+
 4. **Build with helm chart(Option 4) - Cloud Native environment**
 
    Following the instructions on [helm chart README](./helm/README.md) to install RustFS on kubernetes cluster.
@@ -207,4 +214,3 @@ top charts.
 [Apache 2.0](https://opensource.org/licenses/Apache-2.0)
 
 **RustFS** is a trademark of RustFS, Inc. All other trademarks are the property of their respective owners.
-

README_ZH.md

@@ -113,12 +113,14 @@ RustFS 是一个使用 Rust（全球最受欢迎的编程语言之一）构建
 
    你也可以使用 Makefile 提供的目标命令以提升便捷性：
 
-   ```bash
-   make docker-buildx                    # 本地构建
-   make docker-buildx-push               # 构建并推送
-   make docker-buildx-version VERSION=v1.0.0  # 构建指定版本
-   make help-docker                      # 显示全部 Docker 相关命令
-   ```
+  ```bash
+  make docker-buildx                    # 本地构建
+  make docker-buildx-push               # 构建并推送
+  make docker-buildx-version VERSION=v1.0.0  # 构建指定版本
+  make help-docker                      # 显示全部 Docker 相关命令
+  ```
+
+   > **提示（macOS 交叉编译）**：macOS 默认的 `ulimit -n` 只有 256，使用 `cargo zigbuild` 或 `./build-rustfs.sh --platform ...` 编译 Linux 目标时容易触发 `ProcessFdQuotaExceeded` 链接错误。脚本会尝试自动提升该限制，如仍提示失败，请在构建前手动执行 `ulimit -n 4096`（或更大的值）。
 
 4. **使用 Helm Chart 部署（方案四）- 云原生环境**
 

build-rustfs.sh

@@ -163,6 +163,35 @@ print_message() {
     echo -e "${color}${message}${NC}"
 }
 
+# Prevent zig/ld from hitting macOS file descriptor defaults during linking
+ensure_file_descriptor_limit() {
+    local required_limit=4096
+    local current_limit
+    current_limit=$(ulimit -Sn 2>/dev/null || echo "")
+
+    if [ -z "$current_limit" ] || [ "$current_limit" = "unlimited" ]; then
+        return
+    fi
+
+    if (( current_limit >= required_limit )); then
+        return
+    fi
+
+    local hard_limit target_limit
+    hard_limit=$(ulimit -Hn 2>/dev/null || echo "")
+    target_limit=$required_limit
+
+    if [ -n "$hard_limit" ] && [ "$hard_limit" != "unlimited" ] && (( hard_limit < required_limit )); then
+        target_limit=$hard_limit
+    fi
+
+    if ulimit -Sn "$target_limit" 2>/dev/null; then
+        print_message $YELLOW "🔧 Increased open file limit from $current_limit to $target_limit to avoid ProcessFdQuotaExceeded"
+    else
+        print_message $YELLOW "⚠️ Unable to raise ulimit -n automatically (current: $current_limit, needed: $required_limit). Please run 'ulimit -n $required_limit' manually before building."
+    fi
+}
+
 # Get version from git
 get_version() {
     if git describe --abbrev=0 --tags >/dev/null 2>&1; then
@@ -570,10 +599,11 @@ main() {
         fi
     fi
 
+    ensure_file_descriptor_limit
+
     # Start build process
     build_rustfs
 }
 
 # Run main function
 main
-

crates/ahm/src/heal/erasure_healer.rs

@@ -49,8 +49,9 @@ impl ErasureSetHealer {
     }
 
     /// execute erasure set heal with resume
+    #[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))]
     pub async fn heal_erasure_set(&self, buckets: &[String], set_disk_id: &str) -> Result<()> {
-        info!("Starting erasure set heal for {} buckets on set disk {}", buckets.len(), set_disk_id);
+        info!("Starting erasure set heal");
 
         // 1. generate or get task id
         let task_id = self.get_or_create_task_id(set_disk_id).await?;
@@ -231,6 +232,7 @@ impl ErasureSetHealer {
 
     /// heal single bucket with resume
     #[allow(clippy::too_many_arguments)]
+    #[tracing::instrument(skip(self, current_object_index, processed_objects, successful_objects, failed_objects, _skipped_objects, resume_manager, checkpoint_manager), fields(bucket = %bucket, bucket_index = bucket_index))]
     async fn heal_bucket_with_resume(
         &self,
         bucket: &str,
@@ -243,7 +245,7 @@ impl ErasureSetHealer {
         resume_manager: &ResumeManager,
         checkpoint_manager: &CheckpointManager,
     ) -> Result<()> {
-        info!(target: "rustfs:ahm:heal_bucket_with_resume" ,"Starting heal for bucket: {} from object index {}", bucket, current_object_index);
+        info!(target: "rustfs:ahm:heal_bucket_with_resume" ,"Starting heal for bucket from object index {}", current_object_index);
 
         // 1. get bucket info
         let _bucket_info = match self.storage.get_bucket_info(bucket).await? {
@@ -273,7 +275,9 @@ impl ErasureSetHealer {
             let object_exists = match self.storage.object_exists(bucket, object).await {
                 Ok(exists) => exists,
                 Err(e) => {
-                    warn!("Failed to check existence of {}/{}: {}, skipping", bucket, object, e);
+                    warn!("Failed to check existence of {}/{}: {}, marking as failed", bucket, object, e);
+                    *failed_objects += 1;
+                    checkpoint_manager.add_failed_object(object.clone()).await?;
                     *current_object_index = obj_idx + 1;
                     continue;
                 }
@@ -363,7 +367,7 @@ impl ErasureSetHealer {
                 let _permit = semaphore
                     .acquire()
                     .await
-                    .map_err(|e| Error::other(format!("Failed to acquire semaphore for bucket heal: {}", e)))?;
+                    .map_err(|e| Error::other(format!("Failed to acquire semaphore for bucket heal: {e}")))?;
 
                 if cancel_token.is_cancelled() {
                     return Err(Error::TaskCancelled);
@@ -461,7 +465,7 @@ impl ErasureSetHealer {
                 let _permit = semaphore
                     .acquire()
                     .await
-                    .map_err(|e| Error::other(format!("Failed to acquire semaphore for object heal: {}", e)))?;
+                    .map_err(|e| Error::other(format!("Failed to acquire semaphore for object heal: {e}")))?;
 
                 match storage.heal_object(&bucket, &object, None, &heal_opts).await {
                     Ok((_result, None)) => {

crates/ahm/src/heal/manager.rs

@@ -22,7 +22,7 @@ use rustfs_ecstore::disk::DiskAPI;
 use rustfs_ecstore::disk::error::DiskError;
 use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::{BinaryHeap, HashMap, HashSet},
     sync::Arc,
     time::{Duration, SystemTime},
 };
@@ -33,6 +33,151 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 
+/// Priority queue wrapper for heal requests
+/// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items
+#[derive(Debug)]
+struct PriorityHealQueue {
+    /// Heap of (priority, sequence, request) tuples
+    heap: BinaryHeap<PriorityQueueItem>,
+    /// Sequence counter for FIFO ordering within same priority
+    sequence: u64,
+    /// Set of request keys to prevent duplicates
+    dedup_keys: HashSet<String>,
+}
+
+/// Wrapper for heap items to implement proper ordering
+#[derive(Debug)]
+struct PriorityQueueItem {
+    priority: HealPriority,
+    sequence: u64,
+    request: HealRequest,
+}
+
+impl Eq for PriorityQueueItem {}
+
+impl PartialEq for PriorityQueueItem {
+    fn eq(&self, other: &Self) -> bool {
+        self.priority == other.priority && self.sequence == other.sequence
+    }
+}
+
+impl Ord for PriorityQueueItem {
+    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
+        // First compare by priority (higher priority first)
+        match self.priority.cmp(&other.priority) {
+            std::cmp::Ordering::Equal => {
+                // If priorities are equal, use sequence for FIFO (lower sequence first)
+                other.sequence.cmp(&self.sequence)
+            }
+            ordering => ordering,
+        }
+    }
+}
+
+impl PartialOrd for PriorityQueueItem {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl PriorityHealQueue {
+    fn new() -> Self {
+        Self {
+            heap: BinaryHeap::new(),
+            sequence: 0,
+            dedup_keys: HashSet::new(),
+        }
+    }
+
+    fn len(&self) -> usize {
+        self.heap.len()
+    }
+
+    fn is_empty(&self) -> bool {
+        self.heap.is_empty()
+    }
+
+    fn push(&mut self, request: HealRequest) -> bool {
+        let key = Self::make_dedup_key(&request);
+
+        // Check for duplicates
+        if self.dedup_keys.contains(&key) {
+            return false; // Duplicate request, don't add
+        }
+
+        self.dedup_keys.insert(key);
+        self.sequence += 1;
+        self.heap.push(PriorityQueueItem {
+            priority: request.priority,
+            sequence: self.sequence,
+            request,
+        });
+        true
+    }
+
+    /// Get statistics about queue contents by priority
+    fn get_priority_stats(&self) -> HashMap<HealPriority, usize> {
+        let mut stats = HashMap::new();
+        for item in &self.heap {
+            *stats.entry(item.priority).or_insert(0) += 1;
+        }
+        stats
+    }
+
+    fn pop(&mut self) -> Option<HealRequest> {
+        self.heap.pop().map(|item| {
+            let key = Self::make_dedup_key(&item.request);
+            self.dedup_keys.remove(&key);
+            item.request
+        })
+    }
+
+    /// Create a deduplication key from a heal request
+    fn make_dedup_key(request: &HealRequest) -> String {
+        match &request.heal_type {
+            HealType::Object {
+                bucket,
+                object,
+                version_id,
+            } => {
+                format!("object:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
+            }
+            HealType::Bucket { bucket } => {
+                format!("bucket:{}", bucket)
+            }
+            HealType::ErasureSet { set_disk_id, .. } => {
+                format!("erasure_set:{}", set_disk_id)
+            }
+            HealType::Metadata { bucket, object } => {
+                format!("metadata:{}:{}", bucket, object)
+            }
+            HealType::MRF { meta_path } => {
+                format!("mrf:{}", meta_path)
+            }
+            HealType::ECDecode {
+                bucket,
+                object,
+                version_id,
+            } => {
+                format!("ecdecode:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
+            }
+        }
+    }
+
+    /// Check if a request with the same key already exists in the queue
+    #[allow(dead_code)]
+    fn contains_key(&self, request: &HealRequest) -> bool {
+        let key = Self::make_dedup_key(request);
+        self.dedup_keys.contains(&key)
+    }
+
+    /// Check if an erasure set heal request for a specific set_disk_id exists
+    fn contains_erasure_set(&self, set_disk_id: &str) -> bool {
+        let key = format!("erasure_set:{}", set_disk_id);
+        self.dedup_keys.contains(&key)
+    }
+}
+
 /// Heal config
 #[derive(Debug, Clone)]
 pub struct HealConfig {
@@ -85,8 +230,8 @@ pub struct HealManager {
     state: Arc<RwLock<HealState>>,
     /// Active heal tasks
     active_heals: Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
-    /// Heal queue
-    heal_queue: Arc<Mutex<VecDeque<HealRequest>>>,
+    /// Heal queue (priority-based)
+    heal_queue: Arc<Mutex<PriorityHealQueue>>,
     /// Storage layer interface
     storage: Arc<dyn HealStorageAPI>,
     /// Cancel token
@@ -103,7 +248,7 @@ impl HealManager {
             config: Arc::new(RwLock::new(config)),
             state: Arc::new(RwLock::new(HealState::default())),
             active_heals: Arc::new(Mutex::new(HashMap::new())),
-            heal_queue: Arc::new(Mutex::new(VecDeque::new())),
+            heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
             storage,
             cancel_token: CancellationToken::new(),
             statistics: Arc::new(RwLock::new(HealStatistics::new())),
@@ -161,17 +306,54 @@ impl HealManager {
         let config = self.config.read().await;
         let mut queue = self.heal_queue.lock().await;
 
-        if queue.len() >= config.queue_size {
+        let queue_len = queue.len();
+        let queue_capacity = config.queue_size;
+
+        if queue_len >= queue_capacity {
             return Err(Error::ConfigurationError {
-                message: "Heal queue is full".to_string(),
+                message: format!("Heal queue is full ({}/{})", queue_len, queue_capacity),
             });
         }
 
+        // Warn when queue is getting full (>80% capacity)
+        let capacity_threshold = (queue_capacity as f64 * 0.8) as usize;
+        if queue_len >= capacity_threshold {
+            warn!(
+                "Heal queue is {}% full ({}/{}). Consider increasing queue size or processing capacity.",
+                (queue_len * 100) / queue_capacity,
+                queue_len,
+                queue_capacity
+            );
+        }
+
         let request_id = request.id.clone();
-        queue.push_back(request);
+        let priority = request.priority;
+
+        // Try to push the request; if it's a duplicate, still return the request_id
+        let is_new = queue.push(request);
+
+        // Log queue statistics periodically (when adding high/urgent priority items)
+        if matches!(priority, HealPriority::High | HealPriority::Urgent) {
+            let stats = queue.get_priority_stats();
+            info!(
+                "Heal queue stats after adding {:?} priority request: total={}, urgent={}, high={}, normal={}, low={}",
+                priority,
+                queue_len + 1,
+                stats.get(&HealPriority::Urgent).unwrap_or(&0),
+                stats.get(&HealPriority::High).unwrap_or(&0),
+                stats.get(&HealPriority::Normal).unwrap_or(&0),
+                stats.get(&HealPriority::Low).unwrap_or(&0)
+            );
+        }
+
         drop(queue);
 
-        info!("Submitted heal request: {}", request_id);
+        if is_new {
+            info!("Submitted heal request: {} with priority: {:?}", request_id, priority);
+        } else {
+            info!("Heal request already queued (duplicate): {}", request_id);
+        }
+
         Ok(request_id)
     }
 
@@ -321,13 +503,7 @@ impl HealManager {
                             let mut skip = false;
                             {
                                 let queue = heal_queue.lock().await;
-                                if queue.iter().any(|req| {
-                                    matches!(
-                                        &req.heal_type,
-                                        crate::heal::task::HealType::ErasureSet { set_disk_id: queued_id, .. }
-                                        if queued_id == &set_disk_id
-                                    )
-                                }) {
+                                if queue.contains_erasure_set(&set_disk_id) {
                                     skip = true;
                                 }
                             }
@@ -358,7 +534,7 @@ impl HealManager {
                                 HealPriority::Normal,
                             );
                             let mut queue = heal_queue.lock().await;
-                            queue.push_back(req);
+                            queue.push(req);
                             info!("Enqueued auto erasure set heal for endpoint: {} (set_disk_id: {})", ep, set_disk_id);
                         }
                     }
@@ -369,8 +545,9 @@ impl HealManager {
     }
 
     /// Process heal queue
+    /// Processes multiple tasks per cycle when capacity allows and queue has high-priority items
     async fn process_heal_queue(
-        heal_queue: &Arc<Mutex<VecDeque<HealRequest>>>,
+        heal_queue: &Arc<Mutex<PriorityHealQueue>>,
         active_heals: &Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
         config: &Arc<RwLock<HealConfig>>,
         statistics: &Arc<RwLock<HealStatistics>>,
@@ -379,51 +556,83 @@ impl HealManager {
         let config = config.read().await;
         let mut active_heals_guard = active_heals.lock().await;
 
-        // check if new heal tasks can be started
-        if active_heals_guard.len() >= config.max_concurrent_heals {
+        // Check if new heal tasks can be started
+        let active_count = active_heals_guard.len();
+        if active_count >= config.max_concurrent_heals {
             return;
         }
 
+        // Calculate how many tasks we can start this cycle
+        let available_slots = config.max_concurrent_heals - active_count;
+
         let mut queue = heal_queue.lock().await;
-        if let Some(request) = queue.pop_front() {
-            let task = Arc::new(HealTask::from_request(request, storage.clone()));
-            let task_id = task.id.clone();
-            active_heals_guard.insert(task_id.clone(), task.clone());
-            drop(active_heals_guard);
-            let active_heals_clone = active_heals.clone();
-            let statistics_clone = statistics.clone();
+        let queue_len = queue.len();
 
-            // start heal task
-            tokio::spawn(async move {
-                info!("Starting heal task: {}", task_id);
-                let result = task.execute().await;
-                match result {
-                    Ok(_) => {
-                        info!("Heal task completed successfully: {}", task_id);
-                    }
-                    Err(e) => {
-                        error!("Heal task failed: {} - {}", task_id, e);
-                    }
-                }
-                let mut active_heals_guard = active_heals_clone.lock().await;
-                if let Some(completed_task) = active_heals_guard.remove(&task_id) {
-                    // update statistics
-                    let mut stats = statistics_clone.write().await;
-                    match completed_task.get_status().await {
-                        HealTaskStatus::Completed => {
-                            stats.update_task_completion(true);
+        if queue_len == 0 {
+            return;
+        }
+
+        // Process multiple tasks if:
+        // 1. We have available slots
+        // 2. Queue is not empty
+        // Prioritize urgent/high priority tasks by processing up to 2 tasks per cycle if available
+        let tasks_to_process = if queue_len > 0 {
+            std::cmp::min(available_slots, std::cmp::min(2, queue_len))
+        } else {
+            0
+        };
+
+        for _ in 0..tasks_to_process {
+            if let Some(request) = queue.pop() {
+                let task_priority = request.priority;
+                let task = Arc::new(HealTask::from_request(request, storage.clone()));
+                let task_id = task.id.clone();
+                active_heals_guard.insert(task_id.clone(), task.clone());
+                let active_heals_clone = active_heals.clone();
+                let statistics_clone = statistics.clone();
+
+                // start heal task
+                tokio::spawn(async move {
+                    info!("Starting heal task: {} with priority: {:?}", task_id, task_priority);
+                    let result = task.execute().await;
+                    match result {
+                        Ok(_) => {
+                            info!("Heal task completed successfully: {}", task_id);
                         }
-                        _ => {
-                            stats.update_task_completion(false);
+                        Err(e) => {
+                            error!("Heal task failed: {} - {}", task_id, e);
                         }
                     }
-                    stats.update_running_tasks(active_heals_guard.len() as u64);
-                }
-            });
+                    let mut active_heals_guard = active_heals_clone.lock().await;
+                    if let Some(completed_task) = active_heals_guard.remove(&task_id) {
+                        // update statistics
+                        let mut stats = statistics_clone.write().await;
+                        match completed_task.get_status().await {
+                            HealTaskStatus::Completed => {
+                                stats.update_task_completion(true);
+                            }
+                            _ => {
+                                stats.update_task_completion(false);
+                            }
+                        }
+                        stats.update_running_tasks(active_heals_guard.len() as u64);
+                    }
+                });
+            } else {
+                break;
+            }
+        }
 
-            // update statistics
-            let mut stats = statistics.write().await;
-            stats.total_tasks += 1;
+        // Update statistics for all started tasks
+        let mut stats = statistics.write().await;
+        stats.total_tasks += tasks_to_process as u64;
+
+        // Log queue status if items remain
+        if !queue.is_empty() {
+            let remaining = queue.len();
+            if remaining > 10 {
+                info!("Heal queue has {} pending requests, {} tasks active", remaining, active_heals_guard.len());
+            }
         }
     }
 }
@@ -438,3 +647,333 @@ impl std::fmt::Debug for HealManager {
             .finish()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType};
+
+    #[test]
+    fn test_priority_queue_ordering() {
+        let mut queue = PriorityHealQueue::new();
+
+        // Add requests with different priorities
+        let low_req = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+
+        let normal_req = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket2".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let high_req = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket3".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::High,
+        );
+
+        let urgent_req = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket4".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Urgent,
+        );
+
+        // Add in random order: low, high, normal, urgent
+        assert!(queue.push(low_req));
+        assert!(queue.push(high_req));
+        assert!(queue.push(normal_req));
+        assert!(queue.push(urgent_req));
+
+        assert_eq!(queue.len(), 4);
+
+        // Should pop in priority order: urgent, high, normal, low
+        let popped1 = queue.pop().unwrap();
+        assert_eq!(popped1.priority, HealPriority::Urgent);
+
+        let popped2 = queue.pop().unwrap();
+        assert_eq!(popped2.priority, HealPriority::High);
+
+        let popped3 = queue.pop().unwrap();
+        assert_eq!(popped3.priority, HealPriority::Normal);
+
+        let popped4 = queue.pop().unwrap();
+        assert_eq!(popped4.priority, HealPriority::Low);
+
+        assert_eq!(queue.len(), 0);
+    }
+
+    #[test]
+    fn test_priority_queue_fifo_same_priority() {
+        let mut queue = PriorityHealQueue::new();
+
+        // Add multiple requests with same priority
+        let req1 = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let req2 = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket2".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let req3 = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket3".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let id1 = req1.id.clone();
+        let id2 = req2.id.clone();
+        let id3 = req3.id.clone();
+
+        assert!(queue.push(req1));
+        assert!(queue.push(req2));
+        assert!(queue.push(req3));
+
+        // Should maintain FIFO order for same priority
+        let popped1 = queue.pop().unwrap();
+        assert_eq!(popped1.id, id1);
+
+        let popped2 = queue.pop().unwrap();
+        assert_eq!(popped2.id, id2);
+
+        let popped3 = queue.pop().unwrap();
+        assert_eq!(popped3.id, id3);
+    }
+
+    #[test]
+    fn test_priority_queue_deduplication() {
+        let mut queue = PriorityHealQueue::new();
+
+        let req1 = HealRequest::new(
+            HealType::Object {
+                bucket: "bucket1".to_string(),
+                object: "object1".to_string(),
+                version_id: None,
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let req2 = HealRequest::new(
+            HealType::Object {
+                bucket: "bucket1".to_string(),
+                object: "object1".to_string(),
+                version_id: None,
+            },
+            HealOptions::default(),
+            HealPriority::High,
+        );
+
+        // First request should be added
+        assert!(queue.push(req1));
+        assert_eq!(queue.len(), 1);
+
+        // Second request with same object should be rejected (duplicate)
+        assert!(!queue.push(req2));
+        assert_eq!(queue.len(), 1);
+    }
+
+    #[test]
+    fn test_priority_queue_contains_erasure_set() {
+        let mut queue = PriorityHealQueue::new();
+
+        let req = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec!["bucket1".to_string()],
+                set_disk_id: "pool_0_set_1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        assert!(queue.push(req));
+        assert!(queue.contains_erasure_set("pool_0_set_1"));
+        assert!(!queue.contains_erasure_set("pool_0_set_2"));
+    }
+
+    #[test]
+    fn test_priority_queue_dedup_key_generation() {
+        // Test different heal types generate different keys
+        let obj_req = HealRequest::new(
+            HealType::Object {
+                bucket: "bucket1".to_string(),
+                object: "object1".to_string(),
+                version_id: None,
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let bucket_req = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let erasure_req = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec!["bucket1".to_string()],
+                set_disk_id: "pool_0_set_1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+
+        let obj_key = PriorityHealQueue::make_dedup_key(&obj_req);
+        let bucket_key = PriorityHealQueue::make_dedup_key(&bucket_req);
+        let erasure_key = PriorityHealQueue::make_dedup_key(&erasure_req);
+
+        // All keys should be different
+        assert_ne!(obj_key, bucket_key);
+        assert_ne!(obj_key, erasure_key);
+        assert_ne!(bucket_key, erasure_key);
+
+        assert!(obj_key.starts_with("object:"));
+        assert!(bucket_key.starts_with("bucket:"));
+        assert!(erasure_key.starts_with("erasure_set:"));
+    }
+
+    #[test]
+    fn test_priority_queue_mixed_priorities_and_types() {
+        let mut queue = PriorityHealQueue::new();
+
+        // Add various requests
+        let requests = vec![
+            (
+                HealType::Object {
+                    bucket: "b1".to_string(),
+                    object: "o1".to_string(),
+                    version_id: None,
+                },
+                HealPriority::Low,
+            ),
+            (
+                HealType::Bucket {
+                    bucket: "b2".to_string(),
+                },
+                HealPriority::Urgent,
+            ),
+            (
+                HealType::ErasureSet {
+                    buckets: vec!["b3".to_string()],
+                    set_disk_id: "pool_0_set_1".to_string(),
+                },
+                HealPriority::Normal,
+            ),
+            (
+                HealType::Object {
+                    bucket: "b4".to_string(),
+                    object: "o4".to_string(),
+                    version_id: None,
+                },
+                HealPriority::High,
+            ),
+        ];
+
+        for (heal_type, priority) in requests {
+            let req = HealRequest::new(heal_type, HealOptions::default(), priority);
+            queue.push(req);
+        }
+
+        assert_eq!(queue.len(), 4);
+
+        // Check they come out in priority order
+        let priorities: Vec<HealPriority> = (0..4).filter_map(|_| queue.pop().map(|r| r.priority)).collect();
+
+        assert_eq!(
+            priorities,
+            vec![
+                HealPriority::Urgent,
+                HealPriority::High,
+                HealPriority::Normal,
+                HealPriority::Low,
+            ]
+        );
+    }
+
+    #[test]
+    fn test_priority_queue_stats() {
+        let mut queue = PriorityHealQueue::new();
+
+        // Add requests with different priorities
+        for _ in 0..3 {
+            queue.push(HealRequest::new(
+                HealType::Bucket {
+                    bucket: format!("bucket-low-{}", queue.len()),
+                },
+                HealOptions::default(),
+                HealPriority::Low,
+            ));
+        }
+
+        for _ in 0..2 {
+            queue.push(HealRequest::new(
+                HealType::Bucket {
+                    bucket: format!("bucket-normal-{}", queue.len()),
+                },
+                HealOptions::default(),
+                HealPriority::Normal,
+            ));
+        }
+
+        queue.push(HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-high".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::High,
+        ));
+
+        let stats = queue.get_priority_stats();
+
+        assert_eq!(*stats.get(&HealPriority::Low).unwrap_or(&0), 3);
+        assert_eq!(*stats.get(&HealPriority::Normal).unwrap_or(&0), 2);
+        assert_eq!(*stats.get(&HealPriority::High).unwrap_or(&0), 1);
+        assert_eq!(*stats.get(&HealPriority::Urgent).unwrap_or(&0), 0);
+    }
+
+    #[test]
+    fn test_priority_queue_is_empty() {
+        let mut queue = PriorityHealQueue::new();
+
+        assert!(queue.is_empty());
+
+        queue.push(HealRequest::new(
+            HealType::Bucket {
+                bucket: "test".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        ));
+
+        assert!(!queue.is_empty());
+
+        queue.pop();
+
+        assert!(queue.is_empty());
+    }
+}

crates/ahm/src/heal/resume.rs

@@ -30,7 +30,7 @@ const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
 /// Helper function to convert Path to &str, returning an error if conversion fails
 fn path_to_str(path: &Path) -> Result<&str> {
     path.to_str()
-        .ok_or_else(|| Error::other(format!("Invalid UTF-8 path: {:?}", path)))
+        .ok_or_else(|| Error::other(format!("Invalid UTF-8 path: {path:?}")))
 }
 
 /// resume state

crates/ahm/src/heal/storage.rs

@@ -180,8 +180,7 @@ impl HealStorageAPI for ECStoreHealStorage {
                             MAX_READ_BYTES, bucket, object
                         );
                         return Err(Error::other(format!(
-                            "Object too large: {} bytes (max: {} bytes) for {}/{}",
-                            n_read, MAX_READ_BYTES, bucket, object
+                            "Object too large: {n_read} bytes (max: {MAX_READ_BYTES} bytes) for {bucket}/{object}"
                         )));
                     }
                 }
@@ -401,13 +400,13 @@ impl HealStorageAPI for ECStoreHealStorage {
         match self.ecstore.get_object_info(bucket, object, &Default::default()).await {
             Ok(_) => Ok(true), // Object exists
             Err(e) => {
-                // Map ObjectNotFound to false, other errors to false as well for safety
+                // Map ObjectNotFound to false, other errors must be propagated!
                 if matches!(e, rustfs_ecstore::error::StorageError::ObjectNotFound(_, _)) {
                     debug!("Object not found: {}/{}", bucket, object);
                     Ok(false)
                 } else {
-                    debug!("Error checking object existence {}/{}: {}", bucket, object, e);
-                    Ok(false) // Treat errors as non-existence to be safe
+                    error!("Error checking object existence {}/{}: {}", bucket, object, e);
+                    Err(Error::other(e))
                 }
             }
         }
@@ -499,7 +498,7 @@ impl HealStorageAPI for ECStoreHealStorage {
         match self
             .ecstore
             .clone()
-            .list_objects_v2(bucket, prefix, None, None, 1000, false, None)
+            .list_objects_v2(bucket, prefix, None, None, 1000, false, None, false)
             .await
         {
             Ok(list_info) => {

crates/ahm/src/heal/task.rs

@@ -51,7 +51,7 @@ pub enum HealType {
 }
 
 /// Heal priority
-#[derive(Debug, Default, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
 pub enum HealPriority {
     /// Low priority
     Low = 0,
@@ -272,6 +272,7 @@ impl HealTask {
         }
     }
 
+    #[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))]
     pub async fn execute(&self) -> Result<()> {
         // update status and timestamps atomically to avoid race conditions
         let now = SystemTime::now();
@@ -285,7 +286,7 @@ impl HealTask {
             *task_start_instant = Some(start_instant);
         }
 
-        info!("Starting heal task: {} with type: {:?}", self.id, self.heal_type);
+        info!("Task started");
 
         let result = match &self.heal_type {
             HealType::Object {
@@ -315,7 +316,7 @@ impl HealTask {
             Ok(_) => {
                 let mut status = self.status.write().await;
                 *status = HealTaskStatus::Completed;
-                info!("Heal task completed successfully: {}", self.id);
+                info!("Task completed successfully");
             }
             Err(Error::TaskCancelled) => {
                 let mut status = self.status.write().await;
@@ -354,8 +355,9 @@ impl HealTask {
     }
 
     // specific heal implementation method
+    #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))]
     async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
-        info!("Healing object: {}/{}", bucket, object);
+        info!("Starting object heal workflow");
 
         // update progress
         {
@@ -365,7 +367,7 @@ impl HealTask {
         }
 
         // Step 1: Check if object exists and get metadata
-        info!("Step 1: Checking object existence and metadata");
+        warn!("Step 1: Checking object existence and metadata");
         self.check_control_flags().await?;
         let object_exists = self.await_with_control(self.storage.object_exists(bucket, object)).await?;
         if !object_exists {
@@ -424,7 +426,7 @@ impl HealTask {
 
                     // If heal failed and remove_corrupted is enabled, delete the corrupted object
                     if self.options.remove_corrupted {
-                        warn!("Removing corrupted object: {}/{}", bucket, object);
+                        info!("Removing corrupted object: {}/{}", bucket, object);
                         if !self.options.dry_run {
                             self.await_with_control(self.storage.delete_object(bucket, object)).await?;
                             info!("Successfully deleted corrupted object: {}/{}", bucket, object);
@@ -447,11 +449,9 @@ impl HealTask {
                 info!("Step 3: Verifying heal result");
                 let object_size = result.object_size as u64;
                 info!(
-                    "Heal completed successfully: {}/{} ({} bytes, {} drives healed)",
-                    bucket,
-                    object,
-                    object_size,
-                    result.after.drives.len()
+                    object_size = object_size,
+                    drives_healed = result.after.drives.len(),
+                    "Heal completed successfully"
                 );
 
                 {
@@ -481,7 +481,7 @@ impl HealTask {
 
                 // If heal failed and remove_corrupted is enabled, delete the corrupted object
                 if self.options.remove_corrupted {
-                    warn!("Removing corrupted object: {}/{}", bucket, object);
+                    info!("Removing corrupted object: {}/{}", bucket, object);
                     if !self.options.dry_run {
                         self.await_with_control(self.storage.delete_object(bucket, object)).await?;
                         info!("Successfully deleted corrupted object: {}/{}", bucket, object);

crates/ahm/src/scanner/data_scanner.rs

@@ -1005,6 +1005,7 @@ impl Scanner {
                             100,   // max_keys - small limit for performance
                             false, // fetch_owner
                             None,  // start_after
+                            false, // incl_deleted
                         )
                         .await
                     {

crates/ahm/tests/lifecycle_cache_test.rs

@@ -144,7 +144,7 @@ async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
     let mut wtxn = lmdb_env.write_txn().unwrap();
     let db = match lmdb_env
         .database_options()
-        .name(&format!("bucket_{}", bucket_name))
+        .name(&format!("bucket_{bucket_name}"))
         .types::<I64<BigEndian>, LifecycleContentCodec>()
         .flags(DatabaseFlags::DUP_SORT)
         //.dup_sort_comparator::<>()
@@ -152,7 +152,7 @@ async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
     {
         Ok(db) => db,
         Err(err) => {
-            panic!("lmdb error: {}", err);
+            panic!("lmdb error: {err}");
         }
     };
     let _ = wtxn.commit();
@@ -199,7 +199,7 @@ async fn upload_test_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
         .await
         .expect("Failed to upload test object");
 
-    println!("object_info1: {:?}", object_info);
+    println!("object_info1: {object_info:?}");
 
     info!("Uploaded test object: {}/{} ({} bytes)", bucket, object, object_info.size);
 }
@@ -456,7 +456,7 @@ mod serial_tests {
                     }
 
                     let object_info = convert_record_to_object_info(record);
-                    println!("object_info2: {:?}", object_info);
+                    println!("object_info2: {object_info:?}");
                     let mod_time = object_info.mod_time.unwrap_or(OffsetDateTime::now_utc());
                     let expiry_time = rustfs_ecstore::bucket::lifecycle::lifecycle::expected_expiry_time(mod_time, 1);
 
@@ -494,9 +494,9 @@ mod serial_tests {
                             type_,
                             object_name,
                         } = &elm.1;
-                        println!("cache row:{} {} {} {:?} {}", ver_no, ver_id, mod_time, type_, object_name);
+                        println!("cache row:{ver_no} {ver_id} {mod_time} {type_:?} {object_name}");
                     }
-                    println!("row:{:?}", row);
+                    println!("row:{row:?}");
                 }
                 //drop(iter);
                 wtxn.commit().unwrap();

crates/ahm/tests/lifecycle_integration_test.rs

@@ -277,11 +277,11 @@ async fn create_test_tier(server: u32) {
     };
     let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
     if let Err(err) = tier_config_mgr.add(args, false).await {
-        println!("tier_config_mgr add failed, e: {:?}", err);
+        println!("tier_config_mgr add failed, e: {err:?}");
         panic!("tier add failed. {err}");
     }
     if let Err(e) = tier_config_mgr.save().await {
-        println!("tier_config_mgr save failed, e: {:?}", e);
+        println!("tier_config_mgr save failed, e: {e:?}");
         panic!("tier save failed");
     }
     println!("Created test tier: COLDTIER44");
@@ -299,7 +299,7 @@ async fn object_exists(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bo
 #[allow(dead_code)]
 async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
     if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
-        println!("oi: {:?}", oi);
+        println!("oi: {oi:?}");
         oi.delete_marker
     } else {
         println!("object_is_delete_marker is error");
@@ -311,7 +311,7 @@ async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &
 #[allow(dead_code)]
 async fn object_is_transitioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
     if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
-        println!("oi: {:?}", oi);
+        println!("oi: {oi:?}");
         !oi.transitioned_object.status.is_empty()
     } else {
         println!("object_is_transitioned is error");

crates/appauth/Cargo.toml

@@ -29,6 +29,7 @@ base64-simd = { workspace = true }
 rsa = { workspace = true }
 serde.workspace = true
 serde_json.workspace = true
+rand.workspace = true
 
 [lints]
 workspace = true

crates/appauth/src/token.rs

@@ -12,11 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use rsa::Pkcs1v15Encrypt;
 use rsa::{
-    RsaPrivateKey, RsaPublicKey,
+    Pkcs1v15Encrypt, RsaPrivateKey, RsaPublicKey,
     pkcs8::{DecodePrivateKey, DecodePublicKey},
-    rand_core::OsRng,
 };
 use serde::{Deserialize, Serialize};
 use std::io::{Error, Result};
@@ -33,8 +31,9 @@ pub struct Token {
 /// Returns the encrypted string processed by base64
 pub fn gencode(token: &Token, key: &str) -> Result<String> {
     let data = serde_json::to_vec(token)?;
+    let mut rng = rand::rng();
     let public_key = RsaPublicKey::from_public_key_pem(key).map_err(Error::other)?;
-    let encrypted_data = public_key.encrypt(&mut OsRng, Pkcs1v15Encrypt, &data).map_err(Error::other)?;
+    let encrypted_data = public_key.encrypt(&mut rng, Pkcs1v15Encrypt, &data).map_err(Error::other)?;
     Ok(base64_simd::URL_SAFE_NO_PAD.encode_to_string(&encrypted_data))
 }
 
@@ -76,9 +75,10 @@ mod tests {
         pkcs8::{EncodePrivateKey, EncodePublicKey, LineEnding},
     };
     use std::time::{SystemTime, UNIX_EPOCH};
+
     #[test]
     fn test_gencode_and_parse() {
-        let mut rng = OsRng;
+        let mut rng = rand::rng();
         let bits = 2048;
         let private_key = RsaPrivateKey::new(&mut rng, bits).expect("Failed to generate private key");
         let public_key = RsaPublicKey::from(&private_key);
@@ -101,7 +101,8 @@ mod tests {
 
     #[test]
     fn test_parse_invalid_token() {
-        let private_key_pem = RsaPrivateKey::new(&mut OsRng, 2048)
+        let mut rng = rand::rng();
+        let private_key_pem = RsaPrivateKey::new(&mut rng, 2048)
             .expect("Failed to generate private key")
             .to_pkcs8_pem(LineEnding::LF)
             .unwrap();

crates/audit/Cargo.toml

@@ -30,7 +30,10 @@ rustfs-targets = { workspace = true }
 rustfs-config = { workspace = true, features = ["audit", "constants"] }
 rustfs-ecstore = { workspace = true }
 chrono = { workspace = true }
+const-str = { workspace = true }
 futures = { workspace = true }
+hashbrown = { workspace = true }
+metrics = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
@@ -39,5 +42,6 @@ tracing = { workspace = true, features = ["std", "attributes"] }
 url = { workspace = true }
 rumqttc = { workspace = true }
 
+
 [lints]
 workspace = true

crates/audit/src/entity.rs

@@ -13,18 +13,10 @@
 // limitations under the License.
 
 use chrono::{DateTime, Utc};
+use hashbrown::HashMap;
 use rustfs_targets::EventName;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use std::collections::HashMap;
-
-/// Trait for types that can be serialized to JSON and have a timestamp
-pub trait LogRecord {
-    /// Serialize the record to a JSON string
-    fn to_json(&self) -> String;
-    /// Get the timestamp of the record
-    fn get_timestamp(&self) -> chrono::DateTime<chrono::Utc>;
-}
 
 /// ObjectVersion represents an object version with key and versionId
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
@@ -36,19 +28,12 @@ pub struct ObjectVersion {
 }
 
 impl ObjectVersion {
-    /// Set the object name (chainable)
-    pub fn set_object_name(&mut self, name: String) -> &mut Self {
-        self.object_name = name;
-        self
-    }
-    /// Set the version ID (chainable)
-    pub fn set_version_id(&mut self, version_id: Option<String>) -> &mut Self {
-        self.version_id = version_id;
-        self
+    pub fn new(object_name: String, version_id: Option<String>) -> Self {
+        Self { object_name, version_id }
     }
 }
 
-/// ApiDetails contains API information for the audit entry
+/// `ApiDetails` contains API information for the audit entry.
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct ApiDetails {
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -79,75 +64,86 @@ pub struct ApiDetails {
     pub time_to_response_in_ns: Option<String>,
 }
 
-impl ApiDetails {
-    /// Set API name (chainable)
-    pub fn set_name(&mut self, name: Option<String>) -> &mut Self {
-        self.name = name;
+/// Builder for `ApiDetails`.
+#[derive(Default, Clone)]
+pub struct ApiDetailsBuilder(pub ApiDetails);
+
+impl ApiDetailsBuilder {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn name(mut self, name: impl Into<String>) -> Self {
+        self.0.name = Some(name.into());
         self
     }
-    /// Set bucket name (chainable)
-    pub fn set_bucket(&mut self, bucket: Option<String>) -> &mut Self {
-        self.bucket = bucket;
+
+    pub fn bucket(mut self, bucket: impl Into<String>) -> Self {
+        self.0.bucket = Some(bucket.into());
         self
     }
-    /// Set object name (chainable)
-    pub fn set_object(&mut self, object: Option<String>) -> &mut Self {
-        self.object = object;
+
+    pub fn object(mut self, object: impl Into<String>) -> Self {
+        self.0.object = Some(object.into());
         self
     }
-    /// Set objects list (chainable)
-    pub fn set_objects(&mut self, objects: Option<Vec<ObjectVersion>>) -> &mut Self {
-        self.objects = objects;
+
+    pub fn objects(mut self, objects: Vec<ObjectVersion>) -> Self {
+        self.0.objects = Some(objects);
         self
     }
-    /// Set status (chainable)
-    pub fn set_status(&mut self, status: Option<String>) -> &mut Self {
-        self.status = status;
+
+    pub fn status(mut self, status: impl Into<String>) -> Self {
+        self.0.status = Some(status.into());
         self
     }
-    /// Set status code (chainable)
-    pub fn set_status_code(&mut self, code: Option<i32>) -> &mut Self {
-        self.status_code = code;
+
+    pub fn status_code(mut self, code: i32) -> Self {
+        self.0.status_code = Some(code);
         self
     }
-    /// Set input bytes (chainable)
-    pub fn set_input_bytes(&mut self, bytes: Option<i64>) -> &mut Self {
-        self.input_bytes = bytes;
+
+    pub fn input_bytes(mut self, bytes: i64) -> Self {
+        self.0.input_bytes = Some(bytes);
         self
     }
-    /// Set output bytes (chainable)
-    pub fn set_output_bytes(&mut self, bytes: Option<i64>) -> &mut Self {
-        self.output_bytes = bytes;
+
+    pub fn output_bytes(mut self, bytes: i64) -> Self {
+        self.0.output_bytes = Some(bytes);
         self
     }
-    /// Set header bytes (chainable)
-    pub fn set_header_bytes(&mut self, bytes: Option<i64>) -> &mut Self {
-        self.header_bytes = bytes;
+
+    pub fn header_bytes(mut self, bytes: i64) -> Self {
+        self.0.header_bytes = Some(bytes);
         self
     }
-    /// Set time to first byte (chainable)
-    pub fn set_time_to_first_byte(&mut self, t: Option<String>) -> &mut Self {
-        self.time_to_first_byte = t;
+
+    pub fn time_to_first_byte(mut self, t: impl Into<String>) -> Self {
+        self.0.time_to_first_byte = Some(t.into());
         self
     }
-    /// Set time to first byte in nanoseconds (chainable)
-    pub fn set_time_to_first_byte_in_ns(&mut self, t: Option<String>) -> &mut Self {
-        self.time_to_first_byte_in_ns = t;
+
+    pub fn time_to_first_byte_in_ns(mut self, t: impl Into<String>) -> Self {
+        self.0.time_to_first_byte_in_ns = Some(t.into());
         self
     }
-    /// Set time to response (chainable)
-    pub fn set_time_to_response(&mut self, t: Option<String>) -> &mut Self {
-        self.time_to_response = t;
+
+    pub fn time_to_response(mut self, t: impl Into<String>) -> Self {
+        self.0.time_to_response = Some(t.into());
         self
     }
-    /// Set time to response in nanoseconds (chainable)
-    pub fn set_time_to_response_in_ns(&mut self, t: Option<String>) -> &mut Self {
-        self.time_to_response_in_ns = t;
+
+    pub fn time_to_response_in_ns(mut self, t: impl Into<String>) -> Self {
+        self.0.time_to_response_in_ns = Some(t.into());
         self
     }
+
+    pub fn build(self) -> ApiDetails {
+        self.0
+    }
 }
 
-/// AuditEntry represents an audit log entry
+/// `AuditEntry` represents an audit log entry.
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct AuditEntry {
     pub version: String,
@@ -155,6 +151,7 @@ pub struct AuditEntry {
     pub deployment_id: Option<String>,
     #[serde(rename = "siteName", skip_serializing_if = "Option::is_none")]
     pub site_name: Option<String>,
+    #[serde(with = "chrono::serde::ts_milliseconds")]
     pub time: DateTime<Utc>,
     pub event: EventName,
     #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
@@ -191,200 +188,130 @@ pub struct AuditEntry {
     pub error: Option<String>,
 }
 
-impl AuditEntry {
-    /// Create a new AuditEntry with required fields
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(
-        version: String,
-        deployment_id: Option<String>,
-        site_name: Option<String>,
-        time: DateTime<Utc>,
-        event: EventName,
-        entry_type: Option<String>,
-        trigger: String,
-        api: ApiDetails,
-    ) -> Self {
-        AuditEntry {
-            version,
-            deployment_id,
-            site_name,
-            time,
+/// Constructor for `AuditEntry`.
+pub struct AuditEntryBuilder(AuditEntry);
+
+impl AuditEntryBuilder {
+    /// Create a new builder with all required fields.
+    pub fn new(version: impl Into<String>, event: EventName, trigger: impl Into<String>, api: ApiDetails) -> Self {
+        Self(AuditEntry {
+            version: version.into(),
+            time: Utc::now(),
             event,
-            entry_type,
-            trigger,
+            trigger: trigger.into(),
             api,
-            remote_host: None,
-            request_id: None,
-            user_agent: None,
-            req_path: None,
-            req_host: None,
-            req_node: None,
-            req_claims: None,
-            req_query: None,
-            req_header: None,
-            resp_header: None,
-            tags: None,
-            access_key: None,
-            parent_user: None,
-            error: None,
-        }
+            ..Default::default()
+        })
     }
 
-    /// Set version (chainable)
-    pub fn set_version(&mut self, version: String) -> &mut Self {
-        self.version = version;
-        self
-    }
-    /// Set deployment ID (chainable)
-    pub fn set_deployment_id(&mut self, id: Option<String>) -> &mut Self {
-        self.deployment_id = id;
-        self
-    }
-    /// Set site name (chainable)
-    pub fn set_site_name(&mut self, name: Option<String>) -> &mut Self {
-        self.site_name = name;
-        self
-    }
-    /// Set time (chainable)
-    pub fn set_time(&mut self, time: DateTime<Utc>) -> &mut Self {
-        self.time = time;
-        self
-    }
-    /// Set event (chainable)
-    pub fn set_event(&mut self, event: EventName) -> &mut Self {
-        self.event = event;
-        self
-    }
-    /// Set entry type (chainable)
-    pub fn set_entry_type(&mut self, entry_type: Option<String>) -> &mut Self {
-        self.entry_type = entry_type;
-        self
-    }
-    /// Set trigger (chainable)
-    pub fn set_trigger(&mut self, trigger: String) -> &mut Self {
-        self.trigger = trigger;
-        self
-    }
-    /// Set API details (chainable)
-    pub fn set_api(&mut self, api: ApiDetails) -> &mut Self {
-        self.api = api;
-        self
-    }
-    /// Set remote host (chainable)
-    pub fn set_remote_host(&mut self, host: Option<String>) -> &mut Self {
-        self.remote_host = host;
-        self
-    }
-    /// Set request ID (chainable)
-    pub fn set_request_id(&mut self, id: Option<String>) -> &mut Self {
-        self.request_id = id;
-        self
-    }
-    /// Set user agent (chainable)
-    pub fn set_user_agent(&mut self, agent: Option<String>) -> &mut Self {
-        self.user_agent = agent;
-        self
-    }
-    /// Set request path (chainable)
-    pub fn set_req_path(&mut self, path: Option<String>) -> &mut Self {
-        self.req_path = path;
-        self
-    }
-    /// Set request host (chainable)
-    pub fn set_req_host(&mut self, host: Option<String>) -> &mut Self {
-        self.req_host = host;
-        self
-    }
-    /// Set request node (chainable)
-    pub fn set_req_node(&mut self, node: Option<String>) -> &mut Self {
-        self.req_node = node;
-        self
-    }
-    /// Set request claims (chainable)
-    pub fn set_req_claims(&mut self, claims: Option<HashMap<String, Value>>) -> &mut Self {
-        self.req_claims = claims;
-        self
-    }
-    /// Set request query (chainable)
-    pub fn set_req_query(&mut self, query: Option<HashMap<String, String>>) -> &mut Self {
-        self.req_query = query;
-        self
-    }
-    /// Set request header (chainable)
-    pub fn set_req_header(&mut self, header: Option<HashMap<String, String>>) -> &mut Self {
-        self.req_header = header;
-        self
-    }
-    /// Set response header (chainable)
-    pub fn set_resp_header(&mut self, header: Option<HashMap<String, String>>) -> &mut Self {
-        self.resp_header = header;
-        self
-    }
-    /// Set tags (chainable)
-    pub fn set_tags(&mut self, tags: Option<HashMap<String, Value>>) -> &mut Self {
-        self.tags = tags;
-        self
-    }
-    /// Set access key (chainable)
-    pub fn set_access_key(&mut self, key: Option<String>) -> &mut Self {
-        self.access_key = key;
-        self
-    }
-    /// Set parent user (chainable)
-    pub fn set_parent_user(&mut self, user: Option<String>) -> &mut Self {
-        self.parent_user = user;
-        self
-    }
-    /// Set error message (chainable)
-    pub fn set_error(&mut self, error: Option<String>) -> &mut Self {
-        self.error = error;
+    // event
+    pub fn version(mut self, version: impl Into<String>) -> Self {
+        self.0.version = version.into();
         self
     }
 
-    /// Build AuditEntry from context or parameters (example, can be extended)
-    pub fn from_context(
-        version: String,
-        deployment_id: Option<String>,
-        time: DateTime<Utc>,
-        event: EventName,
-        trigger: String,
-        api: ApiDetails,
-        tags: Option<HashMap<String, Value>>,
-    ) -> Self {
-        AuditEntry {
-            version,
-            deployment_id,
-            site_name: None,
-            time,
-            event,
-            entry_type: None,
-            trigger,
-            api,
-            remote_host: None,
-            request_id: None,
-            user_agent: None,
-            req_path: None,
-            req_host: None,
-            req_node: None,
-            req_claims: None,
-            req_query: None,
-            req_header: None,
-            resp_header: None,
-            tags,
-            access_key: None,
-            parent_user: None,
-            error: None,
-        }
-    }
-}
-
-impl LogRecord for AuditEntry {
-    /// Serialize AuditEntry to JSON string
-    fn to_json(&self) -> String {
-        serde_json::to_string(self).unwrap_or_else(|_| String::from("{}"))
-    }
-    /// Get the timestamp of the audit entry
-    fn get_timestamp(&self) -> DateTime<Utc> {
-        self.time
+    pub fn event(mut self, event: EventName) -> Self {
+        self.0.event = event;
+        self
+    }
+
+    pub fn api(mut self, api_details: ApiDetails) -> Self {
+        self.0.api = api_details;
+        self
+    }
+
+    pub fn deployment_id(mut self, id: impl Into<String>) -> Self {
+        self.0.deployment_id = Some(id.into());
+        self
+    }
+
+    pub fn site_name(mut self, name: impl Into<String>) -> Self {
+        self.0.site_name = Some(name.into());
+        self
+    }
+
+    pub fn time(mut self, time: DateTime<Utc>) -> Self {
+        self.0.time = time;
+        self
+    }
+
+    pub fn entry_type(mut self, entry_type: impl Into<String>) -> Self {
+        self.0.entry_type = Some(entry_type.into());
+        self
+    }
+
+    pub fn remote_host(mut self, host: impl Into<String>) -> Self {
+        self.0.remote_host = Some(host.into());
+        self
+    }
+
+    pub fn request_id(mut self, id: impl Into<String>) -> Self {
+        self.0.request_id = Some(id.into());
+        self
+    }
+
+    pub fn user_agent(mut self, agent: impl Into<String>) -> Self {
+        self.0.user_agent = Some(agent.into());
+        self
+    }
+
+    pub fn req_path(mut self, path: impl Into<String>) -> Self {
+        self.0.req_path = Some(path.into());
+        self
+    }
+
+    pub fn req_host(mut self, host: impl Into<String>) -> Self {
+        self.0.req_host = Some(host.into());
+        self
+    }
+
+    pub fn req_node(mut self, node: impl Into<String>) -> Self {
+        self.0.req_node = Some(node.into());
+        self
+    }
+
+    pub fn req_claims(mut self, claims: HashMap<String, Value>) -> Self {
+        self.0.req_claims = Some(claims);
+        self
+    }
+
+    pub fn req_query(mut self, query: HashMap<String, String>) -> Self {
+        self.0.req_query = Some(query);
+        self
+    }
+
+    pub fn req_header(mut self, header: HashMap<String, String>) -> Self {
+        self.0.req_header = Some(header);
+        self
+    }
+
+    pub fn resp_header(mut self, header: HashMap<String, String>) -> Self {
+        self.0.resp_header = Some(header);
+        self
+    }
+
+    pub fn tags(mut self, tags: HashMap<String, Value>) -> Self {
+        self.0.tags = Some(tags);
+        self
+    }
+
+    pub fn access_key(mut self, key: impl Into<String>) -> Self {
+        self.0.access_key = Some(key.into());
+        self
+    }
+
+    pub fn parent_user(mut self, user: impl Into<String>) -> Self {
+        self.0.parent_user = Some(user.into());
+        self
+    }
+
+    pub fn error(mut self, error: impl Into<String>) -> Self {
+        self.0.error = Some(error.into());
+        self
+    }
+
+    /// Construct the final `AuditEntry`.
+    pub fn build(self) -> AuditEntry {
+        self.0
     }
 }

crates/audit/src/error.rs

@@ -21,7 +21,7 @@ pub type AuditResult<T> = Result<T, AuditError>;
 #[derive(Error, Debug)]
 pub enum AuditError {
     #[error("Configuration error: {0}")]
-    Configuration(String),
+    Configuration(String, #[source] Option<Box<dyn std::error::Error + Send + Sync>>),
 
     #[error("config not loaded")]
     ConfigNotLoaded,
@@ -35,11 +35,14 @@ pub enum AuditError {
     #[error("System already initialized")]
     AlreadyInitialized,
 
+    #[error("Storage not available: {0}")]
+    StorageNotAvailable(String),
+
     #[error("Failed to save configuration: {0}")]
-    SaveConfig(String),
+    SaveConfig(#[source] Box<dyn std::error::Error + Send + Sync>),
 
     #[error("Failed to load configuration: {0}")]
-    LoadConfig(String),
+    LoadConfig(#[source] Box<dyn std::error::Error + Send + Sync>),
 
     #[error("Serialization error: {0}")]
     Serialization(#[from] serde_json::Error),
@@ -49,7 +52,4 @@ pub enum AuditError {
 
     #[error("Join error: {0}")]
     Join(#[from] tokio::task::JoinError),
-
-    #[error("Server storage not initialized: {0}")]
-    ServerNotInitialized(String),
 }

crates/audit/src/global.rs

@@ -15,7 +15,7 @@
 use crate::{AuditEntry, AuditResult, AuditSystem};
 use rustfs_ecstore::config::Config;
 use std::sync::{Arc, OnceLock};
-use tracing::{error, warn};
+use tracing::{debug, error, trace, warn};
 
 /// Global audit system instance
 static AUDIT_SYSTEM: OnceLock<Arc<AuditSystem>> = OnceLock::new();
@@ -30,6 +30,19 @@ pub fn audit_system() -> Option<Arc<AuditSystem>> {
     AUDIT_SYSTEM.get().cloned()
 }
 
+/// A helper macro for executing closures if the global audit system is initialized.
+/// If not initialized, log a warning and return `Ok(())`.
+macro_rules! with_audit_system {
+    ($async_closure:expr) => {
+        if let Some(system) = audit_system() {
+            (async move { $async_closure(system).await }).await
+        } else {
+            warn!("Audit system not initialized, operation skipped.");
+            Ok(())
+        }
+    };
+}
+
 /// Start the global audit system with configuration
 pub async fn start_audit_system(config: Config) -> AuditResult<()> {
     let system = init_audit_system();
@@ -38,32 +51,17 @@ pub async fn start_audit_system(config: Config) -> AuditResult<()> {
 
 /// Stop the global audit system
 pub async fn stop_audit_system() -> AuditResult<()> {
-    if let Some(system) = audit_system() {
-        system.close().await
-    } else {
-        warn!("Audit system not initialized, cannot stop");
-        Ok(())
-    }
+    with_audit_system!(|system: Arc<AuditSystem>| async move { system.close().await })
 }
 
 /// Pause the global audit system
 pub async fn pause_audit_system() -> AuditResult<()> {
-    if let Some(system) = audit_system() {
-        system.pause().await
-    } else {
-        warn!("Audit system not initialized, cannot pause");
-        Ok(())
-    }
+    with_audit_system!(|system: Arc<AuditSystem>| async move { system.pause().await })
 }
 
 /// Resume the global audit system
 pub async fn resume_audit_system() -> AuditResult<()> {
-    if let Some(system) = audit_system() {
-        system.resume().await
-    } else {
-        warn!("Audit system not initialized, cannot resume");
-        Ok(())
-    }
+    with_audit_system!(|system: Arc<AuditSystem>| async move { system.resume().await })
 }
 
 /// Dispatch an audit log entry to all targets
@@ -72,23 +70,23 @@ pub async fn dispatch_audit_log(entry: Arc<AuditEntry>) -> AuditResult<()> {
         if system.is_running().await {
             system.dispatch(entry).await
         } else {
-            // System not running, just drop the log entry without error
+            // The system is initialized but not running (for example, it is suspended). Silently discard log entries based on original logic.
+            // For debugging purposes, it can be useful to add a trace log here.
+            trace!("Audit system is not running, dropping audit entry.");
             Ok(())
         }
     } else {
-        // System not initialized, just drop the log entry without error
+        // The system is not initialized at all. This is a more important state.
+        // It might be better to return an error or log a warning.
+        debug!("Audit system not initialized, dropping audit entry.");
+        // If this should be a hard failure, you can return Err(AuditError::NotInitialized("..."))
         Ok(())
     }
 }
 
 /// Reload the global audit system configuration
 pub async fn reload_audit_config(config: Config) -> AuditResult<()> {
-    if let Some(system) = audit_system() {
-        system.reload_config(config).await
-    } else {
-        warn!("Audit system not initialized, cannot reload config");
-        Ok(())
-    }
+    with_audit_system!(|system: Arc<AuditSystem>| async move { system.reload_config(config).await })
 }
 
 /// Check if the global audit system is running

crates/audit/src/lib.rs

@@ -25,7 +25,7 @@ pub mod observability;
 pub mod registry;
 pub mod system;
 
-pub use entity::{ApiDetails, AuditEntry, LogRecord, ObjectVersion};
+pub use entity::{ApiDetails, AuditEntry, ObjectVersion};
 pub use error::{AuditError, AuditResult};
 pub use global::*;
 pub use observability::{AuditMetrics, AuditMetricsReport, PerformanceValidation};

crates/audit/src/observability.rs

@@ -21,12 +21,47 @@
 //! - Error rate monitoring
 //! - Queue depth monitoring
 
+use metrics::{counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, OnceLock};
 use std::time::{Duration, Instant};
 use tokio::sync::RwLock;
 use tracing::info;
 
+const RUSTFS_AUDIT_METRICS_NAMESPACE: &str = "rustfs.audit.";
+
+const M_AUDIT_EVENTS_TOTAL: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "events.total");
+const M_AUDIT_EVENTS_FAILED: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "events.failed");
+const M_AUDIT_DISPATCH_NS: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "dispatch.ns");
+const M_AUDIT_EPS: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "eps");
+const M_AUDIT_TARGET_OPS: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "target.ops");
+const M_AUDIT_CONFIG_RELOADS: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "config.reloads");
+const M_AUDIT_SYSTEM_STARTS: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "system.starts");
+
+const L_RESULT: &str = "result";
+const L_STATUS: &str = "status";
+
+const V_SUCCESS: &str = "success";
+const V_FAILURE: &str = "failure";
+
+/// One-time registration of indicator meta information
+/// This function ensures that metric descriptors are registered only once.
+pub fn init_observability_metrics() {
+    static METRICS_DESC_INIT: OnceLock<()> = OnceLock::new();
+    METRICS_DESC_INIT.get_or_init(|| {
+        // Event/Time-consuming
+        describe_counter!(M_AUDIT_EVENTS_TOTAL, "Total audit events (labeled by result).");
+        describe_counter!(M_AUDIT_EVENTS_FAILED, "Total failed audit events.");
+        describe_histogram!(M_AUDIT_DISPATCH_NS, "Dispatch time per event (ns).");
+        describe_gauge!(M_AUDIT_EPS, "Events per second since last reset.");
+
+        // Target operation/system event
+        describe_counter!(M_AUDIT_TARGET_OPS, "Total target operations (labeled by status).");
+        describe_counter!(M_AUDIT_CONFIG_RELOADS, "Total configuration reloads.");
+        describe_counter!(M_AUDIT_SYSTEM_STARTS, "Total system starts.");
+    });
+}
+
 /// Metrics collector for audit system observability
 #[derive(Debug)]
 pub struct AuditMetrics {
@@ -56,6 +91,7 @@ impl Default for AuditMetrics {
 impl AuditMetrics {
     /// Creates a new metrics collector
     pub fn new() -> Self {
+        init_observability_metrics();
         Self {
             total_events_processed: AtomicU64::new(0),
             total_events_failed: AtomicU64::new(0),
@@ -68,11 +104,28 @@ impl AuditMetrics {
         }
     }
 
+    // Suggestion: Call this auxiliary function in the existing "Successful Event Recording" method body to complete the instrumentation
+    #[inline]
+    fn emit_event_success_metrics(&self, dispatch_time: Duration) {
+        // count + histogram
+        counter!(M_AUDIT_EVENTS_TOTAL, L_RESULT => V_SUCCESS).increment(1);
+        histogram!(M_AUDIT_DISPATCH_NS).record(dispatch_time.as_nanos() as f64);
+    }
+
+    // Suggestion: Call this auxiliary function in the existing "Failure Event Recording" method body to complete the instrumentation
+    #[inline]
+    fn emit_event_failure_metrics(&self, dispatch_time: Duration) {
+        counter!(M_AUDIT_EVENTS_TOTAL, L_RESULT => V_FAILURE).increment(1);
+        counter!(M_AUDIT_EVENTS_FAILED).increment(1);
+        histogram!(M_AUDIT_DISPATCH_NS).record(dispatch_time.as_nanos() as f64);
+    }
+
     /// Records a successful event dispatch
     pub fn record_event_success(&self, dispatch_time: Duration) {
         self.total_events_processed.fetch_add(1, Ordering::Relaxed);
         self.total_dispatch_time_ns
             .fetch_add(dispatch_time.as_nanos() as u64, Ordering::Relaxed);
+        self.emit_event_success_metrics(dispatch_time);
     }
 
     /// Records a failed event dispatch
@@ -80,27 +133,32 @@ impl AuditMetrics {
         self.total_events_failed.fetch_add(1, Ordering::Relaxed);
         self.total_dispatch_time_ns
             .fetch_add(dispatch_time.as_nanos() as u64, Ordering::Relaxed);
+        self.emit_event_failure_metrics(dispatch_time);
     }
 
     /// Records a successful target operation
     pub fn record_target_success(&self) {
         self.target_success_count.fetch_add(1, Ordering::Relaxed);
+        counter!(M_AUDIT_TARGET_OPS, L_STATUS => V_SUCCESS).increment(1);
     }
 
     /// Records a failed target operation
     pub fn record_target_failure(&self) {
         self.target_failure_count.fetch_add(1, Ordering::Relaxed);
+        counter!(M_AUDIT_TARGET_OPS, L_STATUS => V_FAILURE).increment(1);
     }
 
     /// Records a configuration reload
     pub fn record_config_reload(&self) {
         self.config_reload_count.fetch_add(1, Ordering::Relaxed);
+        counter!(M_AUDIT_CONFIG_RELOADS).increment(1);
         info!("Audit configuration reloaded");
     }
 
     /// Records a system start
     pub fn record_system_start(&self) {
         self.system_start_count.fetch_add(1, Ordering::Relaxed);
+        counter!(M_AUDIT_SYSTEM_STARTS).increment(1);
         info!("Audit system started");
     }
 
@@ -110,11 +168,14 @@ impl AuditMetrics {
         let elapsed = reset_time.elapsed();
         let total_events = self.total_events_processed.load(Ordering::Relaxed) + self.total_events_failed.load(Ordering::Relaxed);
 
-        if elapsed.as_secs_f64() > 0.0 {
+        let eps = if elapsed.as_secs_f64() > 0.0 {
             total_events as f64 / elapsed.as_secs_f64()
         } else {
             0.0
-        }
+        };
+        // EPS is reported in gauge
+        gauge!(M_AUDIT_EPS).set(eps);
+        eps
     }
 
     /// Gets the average dispatch latency in milliseconds
@@ -166,6 +227,8 @@ impl AuditMetrics {
         let mut reset_time = self.last_reset_time.write().await;
         *reset_time = Instant::now();
 
+        // Reset EPS to zero after reset
+        gauge!(M_AUDIT_EPS).set(0.0);
         info!("Audit metrics reset");
     }
 

crates/audit/src/registry.rs

@@ -14,6 +14,7 @@
 
 use crate::{AuditEntry, AuditError, AuditResult};
 use futures::{StreamExt, stream::FuturesUnordered};
+use hashbrown::{HashMap, HashSet};
 use rustfs_config::{
     DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR,
     MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE,
@@ -25,7 +26,6 @@ use rustfs_targets::{
     Target, TargetError,
     target::{ChannelTargetType, TargetType, mqtt::MQTTArgs, webhook::WebhookArgs},
 };
-use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::Duration;
 use tracing::{debug, error, info, warn};
@@ -251,7 +251,7 @@ impl AuditRegistry {
             sections.extend(successes_by_section.keys().cloned());
 
             for section_name in sections {
-                let mut section_map: HashMap<String, KVS> = HashMap::new();
+                let mut section_map: std::collections::HashMap<String, KVS> = std::collections::HashMap::new();
 
                 // The default entry (if present) is written back to `_`
                 if let Some(default_cfg) = section_defaults.get(&section_name) {
@@ -277,7 +277,7 @@ impl AuditRegistry {
 
             // 7. Save the new configuration to the system
             let Some(store) = rustfs_ecstore::new_object_layer_fn() else {
-                return Err(AuditError::ServerNotInitialized(
+                return Err(AuditError::StorageNotAvailable(
                     "Failed to save target configuration: server storage not initialized".to_string(),
                 ));
             };
@@ -286,7 +286,7 @@ impl AuditRegistry {
                 Ok(_) => info!("New audit configuration saved to system successfully"),
                 Err(e) => {
                     error!(error = %e, "Failed to save new audit configuration");
-                    return Err(AuditError::SaveConfig(e.to_string()));
+                    return Err(AuditError::SaveConfig(Box::new(e)));
                 }
             }
         }

crates/audit/src/system.rs

@@ -59,7 +59,7 @@ impl AuditSystem {
 
     /// Starts the audit system with the given configuration
     pub async fn start(&self, config: Config) -> AuditResult<()> {
-        let mut state = self.state.write().await;
+        let state = self.state.write().await;
 
         match *state {
             AuditSystemState::Running => {
@@ -72,7 +72,6 @@ impl AuditSystem {
             _ => {}
         }
 
-        *state = AuditSystemState::Starting;
         drop(state);
 
         info!("Starting audit system");
@@ -90,6 +89,17 @@ impl AuditSystem {
         let mut registry = self.registry.lock().await;
         match registry.create_targets_from_config(&config).await {
             Ok(targets) => {
+                if targets.is_empty() {
+                    info!("No enabled audit targets found, keeping audit system stopped");
+                    drop(registry);
+                    return Ok(());
+                }
+
+                {
+                    let mut state = self.state.write().await;
+                    *state = AuditSystemState::Starting;
+                }
+
                 info!(target_count = targets.len(), "Created audit targets successfully");
 
                 // Initialize all targets
@@ -146,7 +156,7 @@ impl AuditSystem {
                 warn!("Audit system is already paused");
                 Ok(())
             }
-            _ => Err(AuditError::Configuration("Cannot pause audit system in current state".to_string())),
+            _ => Err(AuditError::Configuration("Cannot pause audit system in current state".to_string(), None)),
         }
     }
 
@@ -164,7 +174,7 @@ impl AuditSystem {
                 warn!("Audit system is already running");
                 Ok(())
             }
-            _ => Err(AuditError::Configuration("Cannot resume audit system in current state".to_string())),
+            _ => Err(AuditError::Configuration("Cannot resume audit system in current state".to_string(), None)),
         }
     }
 
@@ -460,7 +470,7 @@ impl AuditSystem {
             info!(target_id = %target_id, "Target enabled");
             Ok(())
         } else {
-            Err(AuditError::Configuration(format!("Target not found: {target_id}")))
+            Err(AuditError::Configuration(format!("Target not found: {target_id}"), None))
         }
     }
 
@@ -473,7 +483,7 @@ impl AuditSystem {
             info!(target_id = %target_id, "Target disabled");
             Ok(())
         } else {
-            Err(AuditError::Configuration(format!("Target not found: {target_id}")))
+            Err(AuditError::Configuration(format!("Target not found: {target_id}"), None))
         }
     }
 
@@ -487,7 +497,7 @@ impl AuditSystem {
             info!(target_id = %target_id, "Target removed");
             Ok(())
         } else {
-            Err(AuditError::Configuration(format!("Target not found: {target_id}")))
+            Err(AuditError::Configuration(format!("Target not found: {target_id}"), None))
         }
     }
 

crates/audit/tests/integration_test.rs

@@ -52,7 +52,7 @@ async fn test_config_parsing_webhook() {
     // We expect this to fail due to server storage not being initialized
     // but the parsing should work correctly
     match result {
-        Err(AuditError::ServerNotInitialized(_)) => {
+        Err(AuditError::StorageNotAvailable(_)) => {
             // This is expected in test environment
         }
         Err(e) => {

crates/audit/tests/performance_test.rs

@@ -73,7 +73,7 @@ async fn test_concurrent_target_creation() {
 
     // Verify it fails with expected error (server not initialized)
     match result {
-        Err(AuditError::ServerNotInitialized(_)) => {
+        Err(AuditError::StorageNotAvailable(_)) => {
             // Expected in test environment
         }
         Err(e) => {
@@ -103,17 +103,17 @@ async fn test_audit_log_dispatch_performance() {
     use std::collections::HashMap;
     let id = 1;
 
-    let mut req_header = HashMap::new();
+    let mut req_header = hashbrown::HashMap::new();
     req_header.insert("authorization".to_string(), format!("Bearer test-token-{id}"));
     req_header.insert("content-type".to_string(), "application/octet-stream".to_string());
 
-    let mut resp_header = HashMap::new();
+    let mut resp_header = hashbrown::HashMap::new();
     resp_header.insert("x-response".to_string(), "ok".to_string());
 
-    let mut tags = HashMap::new();
+    let mut tags = hashbrown::HashMap::new();
     tags.insert(format!("tag-{id}"), json!("sample"));
 
-    let mut req_query = HashMap::new();
+    let mut req_query = hashbrown::HashMap::new();
     req_query.insert("id".to_string(), id.to_string());
 
     let api_details = ApiDetails {

crates/audit/tests/system_integration_test.rs

@@ -35,7 +35,7 @@ async fn test_complete_audit_system_lifecycle() {
 
     // Should fail in test environment but state handling should work
     match start_result {
-        Err(AuditError::ServerNotInitialized(_)) => {
+        Err(AuditError::StorageNotAvailable(_)) => {
             // Expected in test environment
             assert_eq!(system.get_state().await, system::AuditSystemState::Stopped);
         }
@@ -168,7 +168,7 @@ async fn test_config_parsing_with_multiple_instances() {
 
     // Should fail due to server storage not initialized, but parsing should work
     match result {
-        Err(AuditError::ServerNotInitialized(_)) => {
+        Err(AuditError::StorageNotAvailable(_)) => {
             // Expected - parsing worked but save failed
         }
         Err(e) => {
@@ -182,48 +182,6 @@ async fn test_config_parsing_with_multiple_instances() {
     }
 }
 
-// #[tokio::test]
-// async fn test_environment_variable_precedence() {
-//     // Test that environment variables override config file settings
-//     // This test validates the ENV > file instance > file default precedence
-//     // Set some test environment variables
-//     std::env::set_var("RUSTFS_AUDIT_WEBHOOK_ENABLE_TEST", "on");
-//     std::env::set_var("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_TEST", "http://env.example.com/audit");
-//     std::env::set_var("RUSTFS_AUDIT_WEBHOOK_AUTH_TOKEN_TEST", "env-token");
-//     let mut registry = AuditRegistry::new();
-//
-//     // Create config that should be overridden by env vars
-//     let mut config = Config(HashMap::new());
-//     let mut webhook_section = HashMap::new();
-//
-//     let mut test_kvs = KVS::new();
-//     test_kvs.insert("enable".to_string(), "off".to_string()); // Should be overridden
-//     test_kvs.insert("endpoint".to_string(), "http://file.example.com/audit".to_string()); // Should be overridden
-//     test_kvs.insert("batch_size".to_string(), "10".to_string()); // Should remain from file
-//     webhook_section.insert("test".to_string(), test_kvs);
-//
-//     config.0.insert("audit_webhook".to_string(), webhook_section);
-//
-//     // Try to create targets - should use env vars for endpoint/enable, file for batch_size
-//     let result = registry.create_targets_from_config(&config).await;
-//     // Clean up env vars
-//     std::env::remove_var("RUSTFS_AUDIT_WEBHOOK_ENABLE_TEST");
-//     std::env::remove_var("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_TEST");
-//     std::env::remove_var("RUSTFS_AUDIT_WEBHOOK_AUTH_TOKEN_TEST");
-//     // Should fail due to server storage, but precedence logic should work
-//     match result {
-//         Err(AuditError::ServerNotInitialized(_)) => {
-//             // Expected - precedence parsing worked but save failed
-//         }
-//         Err(e) => {
-//             println!("Environment precedence test error: {}", e);
-//         }
-//         Ok(_) => {
-//             println!("Unexpected success in environment precedence test");
-//         }
-//     }
-// }
-
 #[test]
 fn test_target_type_validation() {
     use rustfs_targets::target::TargetType;
@@ -315,19 +273,18 @@ fn create_sample_audit_entry_with_id(id: u32) -> AuditEntry {
     use chrono::Utc;
     use rustfs_targets::EventName;
     use serde_json::json;
-    use std::collections::HashMap;
 
-    let mut req_header = HashMap::new();
+    let mut req_header = hashbrown::HashMap::new();
     req_header.insert("authorization".to_string(), format!("Bearer test-token-{id}"));
     req_header.insert("content-type".to_string(), "application/octet-stream".to_string());
 
-    let mut resp_header = HashMap::new();
+    let mut resp_header = hashbrown::HashMap::new();
     resp_header.insert("x-response".to_string(), "ok".to_string());
 
-    let mut tags = HashMap::new();
+    let mut tags = hashbrown::HashMap::new();
     tags.insert(format!("tag-{id}"), json!("sample"));
 
-    let mut req_query = HashMap::new();
+    let mut req_query = hashbrown::HashMap::new();
     req_query.insert("id".to_string(), id.to_string());
 
     let api_details = ApiDetails {

crates/config/src/constants/app.rs

@@ -145,7 +145,7 @@ pub const DEFAULT_LOG_ROTATION_TIME: &str = "hour";
 /// It is used to keep the logs of the application.
 /// Default value: 30
 /// Environment variable: RUSTFS_OBS_LOG_KEEP_FILES
-pub const DEFAULT_LOG_KEEP_FILES: u16 = 30;
+pub const DEFAULT_LOG_KEEP_FILES: usize = 30;
 
 /// Default log local logging enabled for rustfs
 /// This is the default log local logging enabled for rustfs.

crates/config/src/constants/profiler.rs

@@ -12,30 +12,39 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+/// Profiler related environment variable names and default values
 pub const ENV_ENABLE_PROFILING: &str = "RUSTFS_ENABLE_PROFILING";
 
 // CPU profiling
 pub const ENV_CPU_MODE: &str = "RUSTFS_PROF_CPU_MODE"; // off|continuous|periodic
+/// Frequency of CPU profiling samples
 pub const ENV_CPU_FREQ: &str = "RUSTFS_PROF_CPU_FREQ";
+/// Interval between CPU profiling sessions (for periodic mode)
 pub const ENV_CPU_INTERVAL_SECS: &str = "RUSTFS_PROF_CPU_INTERVAL_SECS";
+/// Duration of each CPU profiling session (for periodic mode)
 pub const ENV_CPU_DURATION_SECS: &str = "RUSTFS_PROF_CPU_DURATION_SECS";
 
-// Memory profiling (jemalloc)
+/// Memory profiling (jemalloc)
 pub const ENV_MEM_PERIODIC: &str = "RUSTFS_PROF_MEM_PERIODIC";
+/// Interval between memory profiling snapshots (for periodic mode)
 pub const ENV_MEM_INTERVAL_SECS: &str = "RUSTFS_PROF_MEM_INTERVAL_SECS";
 
-// Output directory
+/// Output directory
 pub const ENV_OUTPUT_DIR: &str = "RUSTFS_PROF_OUTPUT_DIR";
 
-// Defaults
+/// Defaults for profiler settings
 pub const DEFAULT_ENABLE_PROFILING: bool = false;
-
+/// CPU profiling
 pub const DEFAULT_CPU_MODE: &str = "off";
+/// Frequency of CPU profiling samples
 pub const DEFAULT_CPU_FREQ: usize = 100;
+/// Interval between CPU profiling sessions (for periodic mode)
 pub const DEFAULT_CPU_INTERVAL_SECS: u64 = 300;
+/// Duration of each CPU profiling session (for periodic mode)
 pub const DEFAULT_CPU_DURATION_SECS: u64 = 60;
-
+/// Memory profiling (jemalloc)
 pub const DEFAULT_MEM_PERIODIC: bool = false;
+/// Interval between memory profiling snapshots (for periodic mode)
 pub const DEFAULT_MEM_INTERVAL_SECS: u64 = 300;
-
+/// Output directory
 pub const DEFAULT_OUTPUT_DIR: &str = ".";

crates/config/src/observability/metrics.rs

@@ -0,0 +1,19 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// Metrics collection interval in milliseconds for system metrics (CPU, memory, disk, network).
+pub const DEFAULT_METRICS_SYSTEM_INTERVAL_MS: u64 = 30000;
+
+/// Environment variable for setting the metrics collection interval for system metrics.
+pub const ENV_OBS_METRICS_SYSTEM_INTERVAL_MS: &str = "RUSTFS_OBS_METRICS_SYSTEM_INTERVAL_MS";

crates/config/src/observability/mod.rs

@@ -14,7 +14,13 @@
 
 // Observability Keys
 
+mod metrics;
+pub use metrics::*;
+
 pub const ENV_OBS_ENDPOINT: &str = "RUSTFS_OBS_ENDPOINT";
+pub const ENV_OBS_TRACE_ENDPOINT: &str = "RUSTFS_OBS_TRACE_ENDPOINT";
+pub const ENV_OBS_METRIC_ENDPOINT: &str = "RUSTFS_OBS_METRIC_ENDPOINT";
+pub const ENV_OBS_LOG_ENDPOINT: &str = "RUSTFS_OBS_LOG_ENDPOINT";
 pub const ENV_OBS_USE_STDOUT: &str = "RUSTFS_OBS_USE_STDOUT";
 pub const ENV_OBS_SAMPLE_RATIO: &str = "RUSTFS_OBS_SAMPLE_RATIO";
 pub const ENV_OBS_METER_INTERVAL: &str = "RUSTFS_OBS_METER_INTERVAL";
@@ -65,6 +71,9 @@ mod tests {
     #[test]
     fn test_env_keys() {
         assert_eq!(ENV_OBS_ENDPOINT, "RUSTFS_OBS_ENDPOINT");
+        assert_eq!(ENV_OBS_TRACE_ENDPOINT, "RUSTFS_OBS_TRACE_ENDPOINT");
+        assert_eq!(ENV_OBS_METRIC_ENDPOINT, "RUSTFS_OBS_METRIC_ENDPOINT");
+        assert_eq!(ENV_OBS_LOG_ENDPOINT, "RUSTFS_OBS_LOG_ENDPOINT");
         assert_eq!(ENV_OBS_USE_STDOUT, "RUSTFS_OBS_USE_STDOUT");
         assert_eq!(ENV_OBS_SAMPLE_RATIO, "RUSTFS_OBS_SAMPLE_RATIO");
         assert_eq!(ENV_OBS_METER_INTERVAL, "RUSTFS_OBS_METER_INTERVAL");

crates/crypto/Cargo.toml

@@ -29,7 +29,7 @@ documentation = "https://docs.rs/rustfs-crypto/latest/rustfs_crypto/"
 workspace = true
 
 [dependencies]
-aes-gcm = { workspace = true, features = ["std"], optional = true }
+aes-gcm = { workspace = true, optional = true }
 argon2 = { workspace = true, features = ["std"], optional = true }
 cfg-if = { workspace = true }
 chacha20poly1305 = { workspace = true, optional = true }

crates/crypto/src/encdec/decrypt.rs

@@ -19,127 +19,37 @@ pub fn decrypt_data(password: &[u8], data: &[u8]) -> Result<Vec<u8>, crate::Erro
     use aes_gcm::{Aes256Gcm, KeyInit as _};
     use chacha20poly1305::ChaCha20Poly1305;
 
-    // 32: salt
-    // 1: id
-    // 12: nonce
     const HEADER_LENGTH: usize = 45;
     if data.len() < HEADER_LENGTH {
         return Err(Error::ErrUnexpectedHeader);
     }
 
-    let (salt, id, nonce) = (&data[..32], ID::try_from(data[32])?, &data[33..45]);
-    let data = &data[HEADER_LENGTH..];
+    let (salt, id, nonce_slice) = (&data[..32], ID::try_from(data[32])?, &data[33..45]);
+    let body = &data[HEADER_LENGTH..];
 
     match id {
         ID::Argon2idChaCHa20Poly1305 => {
             let key = id.get_key(password, salt)?;
-            decrypt(ChaCha20Poly1305::new_from_slice(&key)?, nonce, data)
+            decrypt(ChaCha20Poly1305::new_from_slice(&key)?, nonce_slice, body)
         }
         _ => {
             let key = id.get_key(password, salt)?;
-            decrypt(Aes256Gcm::new_from_slice(&key)?, nonce, data)
+            decrypt(Aes256Gcm::new_from_slice(&key)?, nonce_slice, body)
         }
     }
 }
 
-// use argon2::{Argon2, PasswordHasher};
-// use argon2::password_hash::{SaltString};
-// use aes_gcm::{Aes256Gcm, Key, Nonce}; // For AES-GCM
-// use chacha20poly1305::{ChaCha20Poly1305, Key as ChaChaKey, Nonce as ChaChaNonce}; // For ChaCha20
-// use pbkdf2::pbkdf2;
-// use sha2::Sha256;
-// use std::io::{self, Read};
-// use thiserror::Error;
-
-// #[derive(Debug, Error)]
-// pub enum DecryptError {
-//     #[error("unexpected header")]
-//     UnexpectedHeader,
-//     #[error("invalid encryption algorithm ID")]
-//     InvalidAlgorithmId,
-//     #[error("IO error")]
-//     Io(#[from] io::Error),
-//     #[error("decryption error")]
-//     DecryptionError,
-// }
-
-// pub fn decrypt_data2<R: Read>(password: &str, mut data: R) -> Result<Vec<u8>, DecryptError> {
-//     // Parse the stream header
-//     let mut hdr = [0u8; 32 + 1 + 8];
-//     if data.read_exact(&mut hdr).is_err() {
-//         return Err(DecryptError::UnexpectedHeader);
-//     }
-
-//     let salt = &hdr[0..32];
-//     let id = hdr[32];
-//     let nonce = &hdr[33..41];
-
-//     let key = match id {
-//         // Argon2id + AES-GCM
-//         0x01 => {
-//             let salt = SaltString::encode_b64(salt).map_err(|_| DecryptError::DecryptionError)?;
-//             let argon2 = Argon2::default();
-//             let hashed_key = argon2.hash_password(password.as_bytes(), &salt)
-//                 .map_err(|_| DecryptError::DecryptionError)?;
-//             hashed_key.hash.unwrap().as_bytes().to_vec()
-//         }
-//         // Argon2id + ChaCha20Poly1305
-//         0x02 => {
-//             let salt = SaltString::encode_b64(salt).map_err(|_| DecryptError::DecryptionError)?;
-//             let argon2 = Argon2::default();
-//             let hashed_key = argon2.hash_password(password.as_bytes(), &salt)
-//                 .map_err(|_| DecryptError::DecryptionError)?;
-//             hashed_key.hash.unwrap().as_bytes().to_vec()
-//         }
-//         // PBKDF2 + AES-GCM
-//         // 0x03 => {
-//         //     let mut key = [0u8; 32];
-//         //     pbkdf2::<Sha256>(password.as_bytes(), salt, 10000, &mut key);
-//         //     key.to_vec()
-//         // }
-//         _ => return Err(DecryptError::InvalidAlgorithmId),
-//     };
-
-//     // Decrypt data using the corresponding cipher
-//     let mut encrypted_data = Vec::new();
-//     data.read_to_end(&mut encrypted_data)?;
-
-//     let plaintext = match id {
-//         0x01 => {
-//             let cipher = Aes256Gcm::new(Key::from_slice(&key));
-//             let nonce = Nonce::from_slice(nonce);
-//             cipher
-//                 .decrypt(nonce, encrypted_data.as_ref())
-//                 .map_err(|_| DecryptError::DecryptionError)?
-//         }
-//         0x02 => {
-//             let cipher = ChaCha20Poly1305::new(ChaChaKey::from_slice(&key));
-//             let nonce = ChaChaNonce::from_slice(nonce);
-//             cipher
-//                 .decrypt(nonce, encrypted_data.as_ref())
-//                 .map_err(|_| DecryptError::DecryptionError)?
-//         }
-//         0x03 => {
-
-//             let cipher = Aes256Gcm::new(Key::from_slice(&key));
-//             let nonce = Nonce::from_slice(nonce);
-//             cipher
-//                 .decrypt(nonce, encrypted_data.as_ref())
-//                 .map_err(|_| DecryptError::DecryptionError)?
-//         }
-//         _ => return Err(DecryptError::InvalidAlgorithmId),
-//     };
-
-//     Ok(plaintext)
-// }
-
 #[cfg(any(test, feature = "crypto"))]
 #[inline]
 fn decrypt<T: aes_gcm::aead::Aead>(stream: T, nonce: &[u8], data: &[u8]) -> Result<Vec<u8>, crate::Error> {
     use crate::error::Error;
-    stream
-        .decrypt(aes_gcm::Nonce::from_slice(nonce), data)
-        .map_err(Error::ErrDecryptFailed)
+    use aes_gcm::AeadCore;
+    use aes_gcm::aead::array::Array;
+    use core::convert::TryFrom;
+
+    let nonce_arr: Array<u8, <T as AeadCore>::NonceSize> =
+        Array::try_from(nonce).map_err(|_| Error::ErrDecryptFailed(aes_gcm::aead::Error))?;
+    stream.decrypt(&nonce_arr, data).map_err(Error::ErrDecryptFailed)
 }
 
 #[cfg(not(any(test, feature = "crypto")))]

crates/crypto/src/encdec/encrypt.rs

@@ -43,7 +43,7 @@ pub fn encrypt_data(password: &[u8], data: &[u8]) -> Result<Vec<u8>, crate::Erro
         if native_aes() {
             encrypt(Aes256Gcm::new_from_slice(&key)?, &salt, id, data)
         } else {
-            encrypt(ChaCha20Poly1305::new_from_slice(&key)?, &salt, id, data)
+            encrypt(chacha20poly1305::ChaCha20Poly1305::new_from_slice(&key)?, &salt, id, data)
         }
     }
 }
@@ -56,16 +56,19 @@ fn encrypt<T: aes_gcm::aead::Aead>(
     data: &[u8],
 ) -> Result<Vec<u8>, crate::Error> {
     use crate::error::Error;
-    use aes_gcm::aead::rand_core::OsRng;
+    use aes_gcm::AeadCore;
+    use aes_gcm::aead::array::Array;
+    use rand::RngCore;
 
-    let nonce = T::generate_nonce(&mut OsRng);
+    let mut nonce: Array<u8, <T as AeadCore>::NonceSize> = Array::default();
+    rand::rng().fill_bytes(&mut nonce);
 
     let encryptor = stream.encrypt(&nonce, data).map_err(Error::ErrEncryptFailed)?;
 
     let mut ciphertext = Vec::with_capacity(salt.len() + 1 + nonce.len() + encryptor.len());
     ciphertext.extend_from_slice(salt);
     ciphertext.push(id as u8);
-    ciphertext.extend_from_slice(nonce.as_slice());
+    ciphertext.extend_from_slice(&nonce);
     ciphertext.extend_from_slice(&encryptor);
 
     Ok(ciphertext)

crates/ecstore/Cargo.toml

@@ -106,6 +106,7 @@ serde_urlencoded.workspace = true
 google-cloud-storage = { workspace = true }
 google-cloud-auth = { workspace = true }
 aws-config = { workspace = true }
+faster-hex = { workspace = true }
 
 [target.'cfg(not(windows))'.dependencies]
 nix = { workspace = true }

crates/ecstore/src/admin_server_info.rs

@@ -34,9 +34,10 @@ use rustfs_protos::{
 };
 use std::{
     collections::{HashMap, HashSet},
-    time::SystemTime,
+    time::{Duration, SystemTime},
 };
 use time::OffsetDateTime;
+use tokio::time::timeout;
 use tonic::Request;
 use tracing::warn;
 
@@ -44,6 +45,8 @@ use shadow_rs::shadow;
 
 shadow!(build);
 
+const SERVER_PING_TIMEOUT: Duration = Duration::from_secs(1);
+
 // pub const ITEM_OFFLINE: &str = "offline";
 // pub const ITEM_INITIALIZING: &str = "initializing";
 // pub const ITEM_ONLINE: &str = "online";
@@ -83,42 +86,45 @@ async fn is_server_resolvable(endpoint: &Endpoint) -> Result<()> {
         endpoint.url.host_str().unwrap(),
         endpoint.url.port().unwrap()
     );
-    let mut fbb = flatbuffers::FlatBufferBuilder::new();
-    let payload = fbb.create_vector(b"hello world");
 
-    let mut builder = PingBodyBuilder::new(&mut fbb);
-    builder.add_payload(payload);
-    let root = builder.finish();
-    fbb.finish(root, None);
+    let ping_task = async {
+        let mut fbb = flatbuffers::FlatBufferBuilder::new();
+        let payload = fbb.create_vector(b"hello world");
 
-    let finished_data = fbb.finished_data();
+        let mut builder = PingBodyBuilder::new(&mut fbb);
+        builder.add_payload(payload);
+        let root = builder.finish();
+        fbb.finish(root, None);
 
-    let decoded_payload = flatbuffers::root::<PingBody>(finished_data);
-    assert!(decoded_payload.is_ok());
+        let finished_data = fbb.finished_data();
 
-    // Create the client
-    let mut client = node_service_time_out_client(&addr)
+        let decoded_payload = flatbuffers::root::<PingBody>(finished_data);
+        assert!(decoded_payload.is_ok());
+
+        let mut client = node_service_time_out_client(&addr)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+
+        let request = Request::new(PingRequest {
+            version: 1,
+            body: bytes::Bytes::copy_from_slice(finished_data),
+        });
+
+        let response: PingResponse = client.ping(request).await?.into_inner();
+
+        let ping_response_body = flatbuffers::root::<PingBody>(&response.body);
+        if let Err(e) = ping_response_body {
+            eprintln!("{e}");
+        } else {
+            println!("ping_resp:body(flatbuffer): {ping_response_body:?}");
+        }
+
+        Ok(())
+    };
+
+    timeout(SERVER_PING_TIMEOUT, ping_task)
         .await
-        .map_err(|err| Error::other(err.to_string()))?;
-
-    // Build the PingRequest
-    let request = Request::new(PingRequest {
-        version: 1,
-        body: bytes::Bytes::copy_from_slice(finished_data),
-    });
-
-    // Send the request and obtain the response
-    let response: PingResponse = client.ping(request).await?.into_inner();
-
-    // Print the response
-    let ping_response_body = flatbuffers::root::<PingBody>(&response.body);
-    if let Err(e) = ping_response_body {
-        eprintln!("{e}");
-    } else {
-        println!("ping_resp:body(flatbuffer): {ping_response_body:?}");
-    }
-
-    Ok(())
+        .map_err(|_| Error::other("server ping timeout"))?
 }
 
 pub async fn get_local_server_property() -> ServerProperties {

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -1105,10 +1105,17 @@ impl TargetClient {
             Err(e) => match e {
                 SdkError::ServiceError(oe) => match oe.into_err() {
                     HeadBucketError::NotFound(_) => Ok(false),
-                    other => Err(other.into()),
+                    other => Err(S3ClientError::new(format!(
+                        "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{other:?}"
+                    ))),
                 },
+                SdkError::DispatchFailure(e) => Err(S3ClientError::new(format!(
+                    "failed to dispatch bucket exists for bucket:{bucket} error:{e:?}"
+                ))),
 
-                _ => Err(e.into()),
+                _ => Err(S3ClientError::new(format!(
+                    "failed to check bucket exists for bucket:{bucket} error:{e:?}"
+                ))),
             },
         }
     }

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -115,10 +115,9 @@ struct ExpiryTask {
 impl ExpiryOp for ExpiryTask {
     fn op_hash(&self) -> u64 {
         let mut hasher = Sha256::new();
-        let _ = hasher.write(format!("{}", self.obj_info.bucket).as_bytes());
-        let _ = hasher.write(format!("{}", self.obj_info.name).as_bytes());
-        hasher.flush();
-        xxh64::xxh64(hasher.clone().finalize().as_slice(), XXHASH_SEED)
+        hasher.update(format!("{}", self.obj_info.bucket).as_bytes());
+        hasher.update(format!("{}", self.obj_info.name).as_bytes());
+        xxh64::xxh64(hasher.finalize().as_slice(), XXHASH_SEED)
     }
 
     fn as_any(&self) -> &dyn Any {
@@ -171,10 +170,9 @@ struct FreeVersionTask(ObjectInfo);
 impl ExpiryOp for FreeVersionTask {
     fn op_hash(&self) -> u64 {
         let mut hasher = Sha256::new();
-        let _ = hasher.write(format!("{}", self.0.transitioned_object.tier).as_bytes());
-        let _ = hasher.write(format!("{}", self.0.transitioned_object.name).as_bytes());
-        hasher.flush();
-        xxh64::xxh64(hasher.clone().finalize().as_slice(), XXHASH_SEED)
+        hasher.update(format!("{}", self.0.transitioned_object.tier).as_bytes());
+        hasher.update(format!("{}", self.0.transitioned_object.name).as_bytes());
+        xxh64::xxh64(hasher.finalize().as_slice(), XXHASH_SEED)
     }
 
     fn as_any(&self) -> &dyn Any {
@@ -191,10 +189,9 @@ struct NewerNoncurrentTask {
 impl ExpiryOp for NewerNoncurrentTask {
     fn op_hash(&self) -> u64 {
         let mut hasher = Sha256::new();
-        let _ = hasher.write(format!("{}", self.bucket).as_bytes());
-        let _ = hasher.write(format!("{}", self.versions[0].object_name).as_bytes());
-        hasher.flush();
-        xxh64::xxh64(hasher.clone().finalize().as_slice(), XXHASH_SEED)
+        hasher.update(format!("{}", self.bucket).as_bytes());
+        hasher.update(format!("{}", self.versions[0].object_name).as_bytes());
+        xxh64::xxh64(hasher.finalize().as_slice(), XXHASH_SEED)
     }
 
     fn as_any(&self) -> &dyn Any {
@@ -415,10 +412,9 @@ struct TransitionTask {
 impl ExpiryOp for TransitionTask {
     fn op_hash(&self) -> u64 {
         let mut hasher = Sha256::new();
-        let _ = hasher.write(format!("{}", self.obj_info.bucket).as_bytes());
-        //let _ = hasher.write(format!("{}", self.obj_info.versions[0].object_name).as_bytes());
-        hasher.flush();
-        xxh64::xxh64(hasher.clone().finalize().as_slice(), XXHASH_SEED)
+        hasher.update(format!("{}", self.obj_info.bucket).as_bytes());
+        // hasher.update(format!("{}", self.obj_info.versions[0].object_name).as_bytes());
+        xxh64::xxh64(hasher.finalize().as_slice(), XXHASH_SEED)
     }
 
     fn as_any(&self) -> &dyn Any {
@@ -480,7 +476,7 @@ impl TransitionState {
             .and_then(|s| s.parse::<i64>().ok())
             .unwrap_or_else(|| std::cmp::min(num_cpus::get() as i64, 16));
         let mut n = max_workers;
-        let tw = 8; //globalILMConfig.getTransitionWorkers(); 
+        let tw = 8; //globalILMConfig.getTransitionWorkers();
         if tw > 0 {
             n = tw;
         }
@@ -760,9 +756,8 @@ pub async fn expire_transitioned_object(
 pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
     let us = Uuid::new_v4().to_string();
     let mut hasher = Sha256::new();
-    let _ = hasher.write(format!("{}/{}", get_global_deployment_id().unwrap_or_default(), bucket).as_bytes());
-    hasher.flush();
-    let hash = rustfs_utils::crypto::hex(hasher.clone().finalize().as_slice());
+    hasher.update(format!("{}/{}", get_global_deployment_id().unwrap_or_default(), bucket).as_bytes());
+    let hash = rustfs_utils::crypto::hex(hasher.finalize().as_slice());
     let obj = format!("{}/{}/{}/{}", &hash[0..16], &us[0..2], &us[2..4], &us);
     Ok(obj)
 }

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -20,7 +20,7 @@
 
 use sha2::{Digest, Sha256};
 use std::any::Any;
-use std::io::{Cursor, Write};
+use std::io::Write;
 use xxhash_rust::xxh64;
 
 use super::bucket_lifecycle_ops::{ExpiryOp, GLOBAL_ExpiryState, TransitionedObject};
@@ -128,10 +128,9 @@ pub struct Jentry {
 impl ExpiryOp for Jentry {
     fn op_hash(&self) -> u64 {
         let mut hasher = Sha256::new();
-        let _ = hasher.write(format!("{}", self.tier_name).as_bytes());
-        let _ = hasher.write(format!("{}", self.obj_name).as_bytes());
-        hasher.flush();
-        xxh64::xxh64(hasher.clone().finalize().as_slice(), XXHASH_SEED)
+        hasher.update(format!("{}", self.tier_name).as_bytes());
+        hasher.update(format!("{}", self.obj_name).as_bytes());
+        xxh64::xxh64(hasher.finalize().as_slice(), XXHASH_SEED)
     }
 
     fn as_any(&self) -> &dyn Any {

crates/ecstore/src/bucket/policy_sys.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{error::BucketMetadataError, metadata_sys::get_bucket_metadata_sys};
-use crate::error::Result;
+use super::metadata_sys::get_bucket_metadata_sys;
+use crate::error::{Result, StorageError};
 use rustfs_policy::policy::{BucketPolicy, BucketPolicyArgs};
-use tracing::warn;
+use tracing::info;
 
 pub struct PolicySys {}
 
@@ -24,9 +24,8 @@ impl PolicySys {
         match Self::get(args.bucket).await {
             Ok(cfg) => return cfg.is_allowed(args),
             Err(err) => {
-                let berr: BucketMetadataError = err.into();
-                if berr != BucketMetadataError::BucketPolicyNotFound {
-                    warn!("config get err {:?}", berr);
+                if err != StorageError::ConfigNotFound {
+                    info!("config get err {:?}", err);
                 }
             }
         }

crates/ecstore/src/bucket/replication/config.rs

@@ -103,6 +103,8 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
                 if filter.test_tags(&object_tags) {
                     rules.push(rule.clone());
                 }
+            } else {
+                rules.push(rule.clone());
             }
         }
 

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -34,7 +34,7 @@ use rustfs_filemeta::{
 };
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, AMZ_TAGGING_DIRECTIVE, CONTENT_ENCODING, HeaderExt as _,
-    RESERVED_METADATA_PREFIX, RESERVED_METADATA_PREFIX_LOWER, RUSTFS_REPLICATION_AUTUAL_OBJECT_SIZE,
+    RESERVED_METADATA_PREFIX, RESERVED_METADATA_PREFIX_LOWER, RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE,
     RUSTFS_REPLICATION_RESET_STATUS, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, headers,
 };
 use rustfs_utils::path::path_join_buf;
@@ -2324,7 +2324,7 @@ async fn replicate_object_with_multipart(
     let mut user_metadata = HashMap::new();
 
     user_metadata.insert(
-        RUSTFS_REPLICATION_AUTUAL_OBJECT_SIZE.to_string(),
+        RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE.to_string(),
         object_info
             .user_defined
             .get(&format!("{RESERVED_METADATA_PREFIX}actual-size"))

crates/ecstore/src/cache_value/metacache_set.rs

@@ -19,7 +19,7 @@ use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_e
 use std::{future::Future, pin::Pin, sync::Arc};
 use tokio::spawn;
 use tokio_util::sync::CancellationToken;
-use tracing::{error, warn};
+use tracing::{error, info, warn};
 
 pub type AgreedFn = Box<dyn Fn(MetaCacheEntry) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
 pub type PartialFn =
@@ -99,7 +99,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                 match disk.walk_dir(wakl_opts, &mut wr).await {
                     Ok(_res) => {}
                     Err(err) => {
-                        error!("walk dir err {:?}", &err);
+                        info!("walk dir err {:?}", &err);
                         need_fallback = true;
                     }
                 }

crates/ecstore/src/data_usage.rs

@@ -277,6 +277,7 @@ pub async fn compute_bucket_usage(store: Arc<ECStore>, bucket_name: &str) -> Res
                 1000,  // max_keys
                 false, // fetch_owner
                 None,  // start_after
+                false, // incl_deleted
             )
             .await?;
 

crates/ecstore/src/disk/error.rs

@@ -140,6 +140,12 @@ pub enum DiskError {
 
     #[error("io error {0}")]
     Io(io::Error),
+
+    #[error("source stalled")]
+    SourceStalled,
+
+    #[error("timeout")]
+    Timeout,
 }
 
 impl DiskError {
@@ -366,6 +372,8 @@ impl Clone for DiskError {
             DiskError::ErasureWriteQuorum => DiskError::ErasureWriteQuorum,
             DiskError::ErasureReadQuorum => DiskError::ErasureReadQuorum,
             DiskError::ShortWrite => DiskError::ShortWrite,
+            DiskError::SourceStalled => DiskError::SourceStalled,
+            DiskError::Timeout => DiskError::Timeout,
         }
     }
 }
@@ -412,6 +420,8 @@ impl DiskError {
             DiskError::ErasureWriteQuorum => 0x25,
             DiskError::ErasureReadQuorum => 0x26,
             DiskError::ShortWrite => 0x27,
+            DiskError::SourceStalled => 0x28,
+            DiskError::Timeout => 0x29,
         }
     }
 
@@ -456,6 +466,8 @@ impl DiskError {
             0x25 => Some(DiskError::ErasureWriteQuorum),
             0x26 => Some(DiskError::ErasureReadQuorum),
             0x27 => Some(DiskError::ShortWrite),
+            0x28 => Some(DiskError::SourceStalled),
+            0x29 => Some(DiskError::Timeout),
             _ => None,
         }
     }

crates/ecstore/src/disk/local.rs

@@ -806,7 +806,7 @@ impl LocalDisk {
         Ok((bytes, modtime))
     }
 
-    async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &Vec<FileInfo>) -> Result<()> {
+    async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &[FileInfo]) -> Result<()> {
         let volume_dir = self.get_bucket_path(volume)?;
         let xlpath = self.get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?;
 
@@ -820,7 +820,7 @@ impl LocalDisk {
 
         fm.unmarshal_msg(&data)?;
 
-        for fi in fis {
+        for fi in fis.iter() {
             let data_dir = match fm.delete_version(fi) {
                 Ok(res) => res,
                 Err(err) => {
@@ -1136,23 +1136,21 @@ impl LocalDisk {
 
             let name = path_join_buf(&[current.as_str(), entry.as_str()]);
 
-            if !dir_stack.is_empty() {
-                if let Some(pop) = dir_stack.last().cloned() {
-                    if pop < name {
-                        out.write_obj(&MetaCacheEntry {
-                            name: pop.clone(),
-                            ..Default::default()
-                        })
-                        .await?;
+            while let Some(pop) = dir_stack.last().cloned()
+                && pop < name
+            {
+                out.write_obj(&MetaCacheEntry {
+                    name: pop.clone(),
+                    ..Default::default()
+                })
+                .await?;
 
-                        if opts.recursive {
-                            if let Err(er) = Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned)).await {
-                                error!("scan_dir err {:?}", er);
-                            }
-                        }
-                        dir_stack.pop();
+                if opts.recursive {
+                    if let Err(er) = Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned)).await {
+                        error!("scan_dir err {:?}", er);
                     }
                 }
+                dir_stack.pop();
             }
 
             let mut meta = MetaCacheEntry {
@@ -2302,7 +2300,6 @@ impl DiskAPI for LocalDisk {
         let buf = match self.read_all_data(volume, &volume_dir, &xl_path).await {
             Ok(res) => res,
             Err(err) => {
-                //
                 if err != DiskError::FileNotFound {
                     return Err(err);
                 }

crates/ecstore/src/notification_sys.rs

@@ -26,9 +26,11 @@ use rustfs_madmin::metrics::RealtimeMetrics;
 use rustfs_madmin::net::NetInfo;
 use rustfs_madmin::{ItemState, ServerProperties};
 use std::collections::hash_map::DefaultHasher;
+use std::future::Future;
 use std::hash::{Hash, Hasher};
 use std::sync::OnceLock;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
+use tokio::time::timeout;
 use tracing::{error, warn};
 
 lazy_static! {
@@ -220,24 +222,21 @@ impl NotificationSys {
 
     pub async fn server_info(&self) -> Vec<ServerProperties> {
         let mut futures = Vec::with_capacity(self.peer_clients.len());
+        let endpoints = get_global_endpoints();
+        let peer_timeout = Duration::from_secs(2);
 
         for client in self.peer_clients.iter() {
+            let endpoints = endpoints.clone();
             futures.push(async move {
                 if let Some(client) = client {
-                    match client.server_info().await {
-                        Ok(info) => info,
-                        Err(_) => ServerProperties {
-                            uptime: SystemTime::now()
-                                .duration_since(*GLOBAL_BOOT_TIME.get().unwrap())
-                                .unwrap_or_default()
-                                .as_secs(),
-                            version: get_commit_id(),
-                            endpoint: client.host.to_string(),
-                            state: ItemState::Offline.to_string().to_owned(),
-                            disks: get_offline_disks(&client.host.to_string(), &get_global_endpoints()),
-                            ..Default::default()
-                        },
-                    }
+                    let host = client.host.to_string();
+                    call_peer_with_timeout(
+                        peer_timeout,
+                        &host,
+                        || client.server_info(),
+                        || offline_server_properties(&host, &endpoints),
+                    )
+                    .await
                 } else {
                     ServerProperties::default()
                 }
@@ -694,6 +693,43 @@ impl NotificationSys {
     }
 }
 
+async fn call_peer_with_timeout<F, Fut>(
+    timeout_dur: Duration,
+    host_label: &str,
+    op: F,
+    fallback: impl FnOnce() -> ServerProperties,
+) -> ServerProperties
+where
+    F: FnOnce() -> Fut,
+    Fut: Future<Output = Result<ServerProperties>> + Send,
+{
+    match timeout(timeout_dur, op()).await {
+        Ok(Ok(info)) => info,
+        Ok(Err(err)) => {
+            warn!("peer {host_label} server_info failed: {err}");
+            fallback()
+        }
+        Err(_) => {
+            warn!("peer {host_label} server_info timed out after {:?}", timeout_dur);
+            fallback()
+        }
+    }
+}
+
+fn offline_server_properties(host: &str, endpoints: &EndpointServerPools) -> ServerProperties {
+    ServerProperties {
+        uptime: SystemTime::now()
+            .duration_since(*GLOBAL_BOOT_TIME.get().unwrap())
+            .unwrap_or_default()
+            .as_secs(),
+        version: get_commit_id(),
+        endpoint: host.to_string(),
+        state: ItemState::Offline.to_string().to_owned(),
+        disks: get_offline_disks(host, endpoints),
+        ..Default::default()
+    }
+}
+
 fn get_offline_disks(offline_host: &str, endpoints: &EndpointServerPools) -> Vec<rustfs_madmin::Disk> {
     let mut offline_disks = Vec::new();
 
@@ -714,3 +750,57 @@ fn get_offline_disks(offline_host: &str, endpoints: &EndpointServerPools) -> Vec
 
     offline_disks
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn build_props(endpoint: &str) -> ServerProperties {
+        ServerProperties {
+            endpoint: endpoint.to_string(),
+            ..Default::default()
+        }
+    }
+
+    #[tokio::test]
+    async fn call_peer_with_timeout_returns_value_when_fast() {
+        let result = call_peer_with_timeout(
+            Duration::from_millis(50),
+            "peer-1",
+            || async { Ok::<_, Error>(build_props("fast")) },
+            || build_props("fallback"),
+        )
+        .await;
+
+        assert_eq!(result.endpoint, "fast");
+    }
+
+    #[tokio::test]
+    async fn call_peer_with_timeout_uses_fallback_on_error() {
+        let result = call_peer_with_timeout(
+            Duration::from_millis(50),
+            "peer-2",
+            || async { Err::<ServerProperties, _>(Error::other("boom")) },
+            || build_props("fallback"),
+        )
+        .await;
+
+        assert_eq!(result.endpoint, "fallback");
+    }
+
+    #[tokio::test]
+    async fn call_peer_with_timeout_uses_fallback_on_timeout() {
+        let result = call_peer_with_timeout(
+            Duration::from_millis(5),
+            "peer-3",
+            || async {
+                tokio::time::sleep(Duration::from_millis(25)).await;
+                Ok::<_, Error>(build_props("slow"))
+            },
+            || build_props("fallback"),
+        )
+        .await;
+
+        assert_eq!(result.endpoint, "fallback");
+    }
+}

crates/ecstore/src/rpc/http_auth.rs

@@ -15,7 +15,7 @@
 use crate::global::get_global_action_cred;
 use base64::Engine as _;
 use base64::engine::general_purpose;
-use hmac::{Hmac, Mac};
+use hmac::{Hmac, KeyInit, Mac};
 use http::HeaderMap;
 use http::HeaderValue;
 use http::Method;

crates/ecstore/src/rpc/remote_disk.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::path::PathBuf;
+use std::{path::PathBuf, time::Duration};
 
 use bytes::Bytes;
 use futures::lock::Mutex;
@@ -40,7 +40,7 @@ use crate::{
 use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_rio::{HttpReader, HttpWriter};
-use tokio::io::AsyncWrite;
+use tokio::{io::AsyncWrite, net::TcpStream, time::timeout};
 use tonic::Request;
 use tracing::info;
 use uuid::Uuid;
@@ -54,6 +54,8 @@ pub struct RemoteDisk {
     endpoint: Endpoint,
 }
 
+const REMOTE_DISK_ONLINE_PROBE_TIMEOUT: Duration = Duration::from_millis(750);
+
 impl RemoteDisk {
     pub async fn new(ep: &Endpoint, _opt: &DiskOption) -> Result<Self> {
         // let root = fs::canonicalize(ep.url.path()).await?;
@@ -83,11 +85,19 @@ impl DiskAPI for RemoteDisk {
 
     #[tracing::instrument(skip(self))]
     async fn is_online(&self) -> bool {
-        // TODO: connection status tracking
-        if node_service_time_out_client(&self.addr).await.is_ok() {
-            return true;
+        let Some(host) = self.endpoint.url.host_str().map(|host| host.to_string()) else {
+            return false;
+        };
+
+        let port = self.endpoint.url.port_or_known_default().unwrap_or(80);
+
+        match timeout(REMOTE_DISK_ONLINE_PROBE_TIMEOUT, TcpStream::connect((host, port))).await {
+            Ok(Ok(stream)) => {
+                drop(stream);
+                true
+            }
+            _ => false,
         }
-        false
     }
 
     #[tracing::instrument(skip(self))]
@@ -957,6 +967,7 @@ impl DiskAPI for RemoteDisk {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use tokio::net::TcpListener;
     use uuid::Uuid;
 
     #[tokio::test]
@@ -1040,6 +1051,58 @@ mod tests {
         assert!(path.to_string_lossy().contains("storage"));
     }
 
+    #[tokio::test]
+    async fn test_remote_disk_is_online_detects_active_listener() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        let url = url::Url::parse(&format!("http://{}:{}/data/rustfs0", addr.ip(), addr.port())).unwrap();
+        let endpoint = Endpoint {
+            url,
+            is_local: false,
+            pool_idx: 0,
+            set_idx: 0,
+            disk_idx: 0,
+        };
+
+        let disk_option = DiskOption {
+            cleanup: false,
+            health_check: false,
+        };
+
+        let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap();
+        assert!(remote_disk.is_online().await);
+
+        drop(listener);
+    }
+
+    #[tokio::test]
+    async fn test_remote_disk_is_online_detects_missing_listener() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let ip = addr.ip();
+        let port = addr.port();
+
+        drop(listener);
+
+        let url = url::Url::parse(&format!("http://{ip}:{port}/data/rustfs0")).unwrap();
+        let endpoint = Endpoint {
+            url,
+            is_local: false,
+            pool_idx: 0,
+            set_idx: 0,
+            disk_idx: 0,
+        };
+
+        let disk_option = DiskOption {
+            cleanup: false,
+            health_check: false,
+        };
+
+        let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap();
+        assert!(!remote_disk.is_online().await);
+    }
+
     #[tokio::test]
     async fn test_remote_disk_disk_id() {
         let url = url::Url::parse("http://remote-server:9000").unwrap();

crates/ecstore/src/set_disk.rs

@@ -68,6 +68,7 @@ use md5::{Digest as Md5Digest, Md5};
 use rand::{Rng, seq::SliceRandom};
 use regex::Regex;
 use rustfs_common::heal_channel::{DriveState, HealChannelPriority, HealItemType, HealOpts, HealScanMode, send_heal_disk};
+use rustfs_config::MI_B;
 use rustfs_filemeta::{
     FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
     RawFileInfo, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, merge_file_meta_versions,
@@ -88,7 +89,7 @@ use s3s::header::X_AMZ_RESTORE;
 use sha2::{Digest, Sha256};
 use std::hash::Hash;
 use std::mem::{self};
-use std::time::SystemTime;
+use std::time::{Instant, SystemTime};
 use std::{
     collections::{HashMap, HashSet},
     io::{Cursor, Write},
@@ -104,15 +105,17 @@ use tokio::{
 use tokio::{
     select,
     sync::mpsc::{self, Sender},
-    time::interval,
+    time::{interval, timeout},
 };
 use tokio_util::sync::CancellationToken;
 use tracing::error;
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
-pub const DEFAULT_READ_BUFFER_SIZE: usize = 1024 * 1024;
+pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024;
 pub const MAX_PARTS_COUNT: usize = 10000;
+const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
+const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
 
 #[derive(Clone, Debug)]
 pub struct SetDisks {
@@ -125,6 +128,23 @@ pub struct SetDisks {
     pub set_index: usize,
     pub pool_index: usize,
     pub format: FormatV3,
+    disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
+}
+
+#[derive(Clone, Debug)]
+struct DiskHealthEntry {
+    last_check: Instant,
+    online: bool,
+}
+
+impl DiskHealthEntry {
+    fn cached_value(&self) -> Option<bool> {
+        if self.last_check.elapsed() <= DISK_HEALTH_CACHE_TTL {
+            Some(self.online)
+        } else {
+            None
+        }
+    }
 }
 
 impl SetDisks {
@@ -150,8 +170,60 @@ impl SetDisks {
             pool_index,
             format,
             set_endpoints,
+            disk_health_cache: Arc::new(RwLock::new(Vec::new())),
         })
     }
+
+    async fn cached_disk_health(&self, index: usize) -> Option<bool> {
+        let cache = self.disk_health_cache.read().await;
+        cache
+            .get(index)
+            .and_then(|entry| entry.as_ref().and_then(|state| state.cached_value()))
+    }
+
+    async fn update_disk_health(&self, index: usize, online: bool) {
+        let mut cache = self.disk_health_cache.write().await;
+        if cache.len() <= index {
+            cache.resize(index + 1, None);
+        }
+        cache[index] = Some(DiskHealthEntry {
+            last_check: Instant::now(),
+            online,
+        });
+    }
+
+    async fn is_disk_online_cached(&self, index: usize, disk: &DiskStore) -> bool {
+        if let Some(online) = self.cached_disk_health(index).await {
+            return online;
+        }
+
+        let disk_clone = disk.clone();
+        let online = timeout(DISK_ONLINE_TIMEOUT, async move { disk_clone.is_online().await })
+            .await
+            .unwrap_or(false);
+        self.update_disk_health(index, online).await;
+        online
+    }
+
+    async fn filter_online_disks(&self, disks: Vec<Option<DiskStore>>) -> (Vec<Option<DiskStore>>, usize) {
+        let mut filtered = Vec::with_capacity(disks.len());
+        let mut online_count = 0;
+
+        for (idx, disk) in disks.into_iter().enumerate() {
+            if let Some(disk_store) = disk {
+                if self.is_disk_online_cached(idx, &disk_store).await {
+                    filtered.push(Some(disk_store));
+                    online_count += 1;
+                } else {
+                    filtered.push(None);
+                }
+            } else {
+                filtered.push(None);
+            }
+        }
+
+        (filtered, online_count)
+    }
     fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String {
         match err {
             LockResult::Timeout => {
@@ -187,25 +259,9 @@ impl SetDisks {
     }
 
     async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
-        let mut disks = self.get_disks_internal().await;
-
-        // TODO: diskinfo filter online
-
-        let mut new_disk = Vec::with_capacity(disks.len());
-
-        for disk in disks.iter() {
-            if let Some(d) = disk {
-                if d.is_online().await {
-                    new_disk.push(disk.clone());
-                }
-            }
-        }
-
-        let mut rng = rand::rng();
-
-        disks.shuffle(&mut rng);
-
-        new_disk
+        let disks = self.get_disks_internal().await;
+        let (filtered, _) = self.filter_online_disks(disks).await;
+        filtered.into_iter().filter(|disk| disk.is_some()).collect()
     }
     async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
         let mut disks = self.get_online_disks().await;
@@ -1260,21 +1316,38 @@ impl SetDisks {
 
         for (i, meta) in metas.iter().enumerate() {
             if !meta.is_valid() {
+                debug!(
+                    index = i,
+                    valid = false,
+                    version_id = ?meta.version_id,
+                    mod_time = ?meta.mod_time,
+                    "find_file_info_in_quorum: skipping invalid meta"
+                );
                 continue;
             }
 
+            debug!(
+                index = i,
+                valid = true,
+                version_id = ?meta.version_id,
+                mod_time = ?meta.mod_time,
+                deleted = meta.deleted,
+                size = meta.size,
+                "find_file_info_in_quorum: inspecting meta"
+            );
+
             let etag_only = mod_time.is_none() && etag.is_some() && meta.get_etag().is_some_and(|v| &v == etag.as_ref().unwrap());
             let mod_valid = mod_time == &meta.mod_time;
 
             if etag_only || mod_valid {
                 for part in meta.parts.iter() {
-                    let _ = hasher.write(format!("part.{}", part.number).as_bytes())?;
-                    let _ = hasher.write(format!("part.{}", part.size).as_bytes())?;
+                    hasher.update(format!("part.{}", part.number).as_bytes());
+                    hasher.update(format!("part.{}", part.size).as_bytes());
                 }
 
                 if !meta.deleted && meta.size != 0 {
-                    let _ = hasher.write(format!("{}+{}", meta.erasure.data_blocks, meta.erasure.parity_blocks).as_bytes())?;
-                    let _ = hasher.write(format!("{:?}", meta.erasure.distribution).as_bytes())?;
+                    hasher.update(format!("{}+{}", meta.erasure.data_blocks, meta.erasure.parity_blocks).as_bytes());
+                    hasher.update(format!("{:?}", meta.erasure.distribution).as_bytes());
                 }
 
                 if meta.is_remote() {
@@ -1285,11 +1358,16 @@ impl SetDisks {
 
                 // TODO: IsCompressed
 
-                hasher.flush()?;
-
                 meta_hashes[i] = Some(hex(hasher.clone().finalize().as_slice()));
 
                 hasher.reset();
+            } else {
+                debug!(
+                    index = i,
+                    etag_only_match = etag_only,
+                    mod_valid_match = mod_valid,
+                    "find_file_info_in_quorum: meta does not match common etag or mod_time, skipping hash calculation"
+                );
             }
         }
 
@@ -1438,7 +1516,7 @@ impl SetDisks {
         object: &str,
         version_id: &str,
         opts: &ReadOptions,
-    ) -> Result<Vec<rustfs_filemeta::FileInfo>> {
+    ) -> Result<Vec<FileInfo>> {
         // Use existing disk selection logic
         let disks = self.disks.read().await;
         let required_reads = self.format.erasure.sets.len();
@@ -2127,11 +2205,11 @@ impl SetDisks {
         // TODO: replicatio
 
         if fi.deleted {
-            if opts.version_id.is_none() || opts.delete_marker {
-                return Err(to_object_err(StorageError::FileNotFound, vec![bucket, object]));
+            return if opts.version_id.is_none() || opts.delete_marker {
+                Err(to_object_err(StorageError::FileNotFound, vec![bucket, object]))
             } else {
-                return Err(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object]));
-            }
+                Err(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object]))
+            };
         }
 
         Ok((oi, write_quorum))
@@ -2159,7 +2237,7 @@ impl SetDisks {
     where
         W: AsyncWrite + Send + Sync + Unpin + 'static,
     {
-        tracing::debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start");
+        debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start");
         let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi);
 
         let total_size = fi.size as usize;
@@ -2184,27 +2262,20 @@ impl SetDisks {
 
         let (last_part_index, last_part_relative_offset) = fi.to_part_offset(end_offset)?;
 
-        tracing::debug!(
+        debug!(
             bucket,
-            object,
-            offset,
-            length,
-            end_offset,
-            part_index,
-            last_part_index,
-            last_part_relative_offset,
-            "Multipart read bounds"
+            object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
         );
 
         let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
 
         let part_indices: Vec<usize> = (part_index..=last_part_index).collect();
-        tracing::debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
+        debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
 
         let mut total_read = 0;
         for current_part in part_indices {
             if total_read == length {
-                tracing::debug!(
+                debug!(
                     bucket,
                     object,
                     total_read,
@@ -2226,7 +2297,7 @@ impl SetDisks {
 
             let read_offset = (part_offset / erasure.block_size) * erasure.shard_size();
 
-            tracing::debug!(
+            debug!(
                 bucket,
                 object,
                 part_index = current_part,
@@ -2281,12 +2352,65 @@ impl SetDisks {
                 return Err(Error::other(format!("not enough disks to read: {errors:?}")));
             }
 
+            // Check if we have missing shards even though we can read successfully
+            // This happens when a node was offline during write and comes back online
+            let total_shards = erasure.data_shards + erasure.parity_shards;
+            let available_shards = nil_count;
+            let missing_shards = total_shards - available_shards;
+
+            info!(
+                bucket,
+                object,
+                part_number,
+                total_shards,
+                available_shards,
+                missing_shards,
+                data_shards = erasure.data_shards,
+                parity_shards = erasure.parity_shards,
+                "Shard availability check"
+            );
+
+            if missing_shards > 0 && available_shards >= erasure.data_shards {
+                // We have missing shards but enough to read - trigger background heal
+                info!(
+                    bucket,
+                    object,
+                    part_number,
+                    missing_shards,
+                    available_shards,
+                    pool_index,
+                    set_index,
+                    "Detected missing shards during read, triggering background heal"
+                );
+                if let Err(e) =
+                    rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
+                        bucket.to_string(),
+                        Some(object.to_string()),
+                        false,
+                        Some(HealChannelPriority::Normal),
+                        Some(pool_index),
+                        Some(set_index),
+                    ))
+                    .await
+                {
+                    warn!(
+                        bucket,
+                        object,
+                        part_number,
+                        error = %e,
+                        "Failed to enqueue heal request for missing shards"
+                    );
+                } else {
+                    warn!(bucket, object, part_number, "Successfully enqueued heal request for missing shards");
+                }
+            }
+
             // debug!(
             //     "read part {} part_offset {},part_length {},part_size {}  ",
             //     part_number, part_offset, part_length, part_size
             // );
             let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await;
-            tracing::debug!(
+            debug!(
                 bucket,
                 object,
                 part_index = current_part,
@@ -2302,7 +2426,7 @@ impl SetDisks {
                     match de_err {
                         DiskError::FileNotFound | DiskError::FileCorrupt => {
                             error!("erasure.decode err 111 {:?}", &de_err);
-                            let _ = rustfs_common::heal_channel::send_heal_request(
+                            if let Err(e) = rustfs_common::heal_channel::send_heal_request(
                                 rustfs_common::heal_channel::create_heal_request_with_options(
                                     bucket.to_string(),
                                     Some(object.to_string()),
@@ -2312,7 +2436,16 @@ impl SetDisks {
                                     Some(set_index),
                                 ),
                             )
-                            .await;
+                            .await
+                            {
+                                warn!(
+                                    bucket,
+                                    object,
+                                    part_number,
+                                    error = %e,
+                                    "Failed to enqueue heal request after decode error"
+                                );
+                            }
                             has_err = false;
                         }
                         _ => {}
@@ -2333,7 +2466,7 @@ impl SetDisks {
 
         // debug!("read end");
 
-        tracing::debug!(bucket, object, total_read, expected_length = length, "Multipart read finished");
+        debug!(bucket, object, total_read, expected_length = length, "Multipart read finished");
 
         Ok(())
     }
@@ -2452,6 +2585,7 @@ impl SetDisks {
         Ok((new_disks, new_infos, healing))
     }
 
+    #[tracing::instrument(skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))]
     async fn heal_object(
         &self,
         bucket: &str,
@@ -2459,10 +2593,7 @@ impl SetDisks {
         version_id: &str,
         opts: &HealOpts,
     ) -> disk::error::Result<(HealResultItem, Option<DiskError>)> {
-        info!(
-            "SetDisks heal_object: bucket={}, object={}, version_id={}, opts={:?}",
-            bucket, object, version_id, opts
-        );
+        info!(?opts, "Starting heal_object");
         let mut result = HealResultItem {
             heal_item_type: HealItemType::Object.to_string(),
             bucket: bucket.to_string(),
@@ -2494,20 +2625,34 @@ impl SetDisks {
             if reuse_existing_lock {
                 None
             } else {
-                let start_time = std::time::Instant::now();
-                let lock_result = self
-                    .fast_lock_manager
-                    .acquire_write_lock(bucket, object, self.locker_owner.as_str())
-                    .await
-                    .map_err(|e| {
-                        let elapsed = start_time.elapsed();
-                        let message = self.format_lock_error(bucket, object, "write", &e);
-                        error!("Failed to acquire write lock for heal operation after {:?}: {}", elapsed, message);
-                        DiskError::other(message)
-                    })?;
-                let elapsed = start_time.elapsed();
-                info!("Successfully acquired write lock for object: {} in {:?}", object, elapsed);
-                Some(lock_result)
+                let mut lock_result = None;
+                for i in 0..3 {
+                    let start_time = Instant::now();
+                    match self
+                        .fast_lock_manager
+                        .acquire_write_lock(bucket, object, self.locker_owner.as_str())
+                        .await
+                    {
+                        Ok(res) => {
+                            let elapsed = start_time.elapsed();
+                            info!(duration = ?elapsed, attempt = i + 1, "Write lock acquired");
+                            lock_result = Some(res);
+                            break;
+                        }
+                        Err(e) => {
+                            let elapsed = start_time.elapsed();
+                            info!(error = ?e, attempt = i + 1, duration = ?elapsed, "Lock acquisition failed, retrying");
+                            if i < 2 {
+                                tokio::time::sleep(Duration::from_millis(50 * (i as u64 + 1))).await;
+                            } else {
+                                let message = self.format_lock_error(bucket, object, "write", &e);
+                                error!("Failed to acquire write lock after retries: {}", message);
+                                return Err(DiskError::other(message));
+                            }
+                        }
+                    }
+                }
+                lock_result
             }
         } else {
             info!("Skipping lock acquisition (no_lock=true)");
@@ -2524,8 +2669,37 @@ impl SetDisks {
 
         let disks = { self.disks.read().await.clone() };
 
-        let (mut parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true).await?;
-        info!("Read file info: parts_metadata.len()={}, errs={:?}", parts_metadata.len(), errs);
+        let (mut parts_metadata, errs) = {
+            let mut retry_count = 0;
+            loop {
+                let (parts, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true).await?;
+
+                // Check if we have enough valid metadata to proceed
+                // If we have too many errors, and we haven't exhausted retries, try again
+                let valid_count = errs.iter().filter(|e| e.is_none()).count();
+                // Simple heuristic: if valid_count is less than expected quorum (e.g. half disks), retry
+                // But we don't know the exact quorum yet. Let's just retry on high error rate if possible.
+                // Actually, read_all_fileinfo shouldn't fail easily.
+                // Let's just retry if we see ANY non-NotFound errors that might be transient (like timeouts)
+
+                let has_transient_error = errs
+                    .iter()
+                    .any(|e| matches!(e, Some(DiskError::SourceStalled) | Some(DiskError::Timeout)));
+
+                if !has_transient_error || retry_count >= 3 {
+                    break (parts, errs);
+                }
+
+                info!(
+                    "read_all_fileinfo encountered transient errors, retrying (attempt {}/3). Errs: {:?}",
+                    retry_count + 1,
+                    errs
+                );
+                tokio::time::sleep(Duration::from_millis(50 * (retry_count as u64 + 1))).await;
+                retry_count += 1;
+            }
+        };
+        info!(parts_count = parts_metadata.len(), ?errs, "File info read complete");
         if DiskError::is_all_not_found(&errs) {
             warn!(
                 "heal_object failed, all obj part not found, bucket: {}, obj: {}, version_id: {}",
@@ -2544,7 +2718,7 @@ impl SetDisks {
             ));
         }
 
-        info!("About to call object_quorum_from_meta with parts_metadata.len()={}", parts_metadata.len());
+        info!(parts_count = parts_metadata.len(), "Initiating quorum check");
         match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count) {
             Ok((read_quorum, _)) => {
                 result.parity_blocks = result.disk_count - read_quorum as usize;
@@ -2569,10 +2743,12 @@ impl SetDisks {
                         )
                         .await?;
 
-                        // info!(
-                        //     "disks_with_all_parts: got available_disks: {:?}, data_errs_by_disk: {:?}, data_errs_by_part: {:?}, latest_meta: {:?}",
-                        //     available_disks, data_errs_by_disk, data_errs_by_part, latest_meta
-                        // );
+                        info!(
+                            "disks_with_all_parts results: available_disks count={}, total_disks={}",
+                            available_disks.iter().filter(|d| d.is_some()).count(),
+                            available_disks.len()
+                        );
+
                         let erasure = if !latest_meta.deleted && !latest_meta.is_remote() {
                             // Initialize erasure coding
                             erasure_coding::Erasure::new(
@@ -2592,10 +2768,7 @@ impl SetDisks {
                         let mut outdate_disks = vec![None; disk_len];
                         let mut disks_to_heal_count = 0;
 
-                        // info!(
-                        //     "errs: {:?}, data_errs_by_disk: {:?}, latest_meta: {:?}",
-                        //     errs, data_errs_by_disk, latest_meta
-                        // );
+                        info!("Checking {} disks for healing needs (bucket={}, object={})", disk_len, bucket, object);
                         for index in 0..available_disks.len() {
                             let (yes, reason) = should_heal_object_on_disk(
                                 &errs[index],
@@ -2603,9 +2776,16 @@ impl SetDisks {
                                 &parts_metadata[index],
                                 &latest_meta,
                             );
+
+                            info!(
+                                "Disk {} heal check: should_heal={}, reason={:?}, err={:?}, endpoint={}",
+                                index, yes, reason, errs[index], self.set_endpoints[index]
+                            );
+
                             if yes {
                                 outdate_disks[index] = disks[index].clone();
                                 disks_to_heal_count += 1;
+                                info!("Disk {} marked for healing (endpoint={})", index, self.set_endpoints[index]);
                             }
 
                             let drive_state = match reason {
@@ -2633,6 +2813,11 @@ impl SetDisks {
                             });
                         }
 
+                        info!(
+                            "Heal check complete: {} disks need healing out of {} total (bucket={}, object={})",
+                            disks_to_heal_count, disk_len, bucket, object
+                        );
+
                         if DiskError::is_all_not_found(&errs) {
                             warn!(
                                 "heal_object failed, all obj part not found, bucket: {}, obj: {}, version_id: {}",
@@ -2667,9 +2852,18 @@ impl SetDisks {
                         );
 
                         if !latest_meta.deleted && disks_to_heal_count > latest_meta.erasure.parity_blocks {
+                            let total_disks = parts_metadata.len();
+                            let healthy_count = total_disks.saturating_sub(disks_to_heal_count);
+                            let required_data = total_disks.saturating_sub(latest_meta.erasure.parity_blocks);
+
                             error!(
-                                "file({} : {}) part corrupt too much, can not to fix, disks_to_heal_count: {}, parity_blocks: {}",
-                                bucket, object, disks_to_heal_count, latest_meta.erasure.parity_blocks
+                                "Data corruption detected for {}/{}: Insufficient healthy shards. Need at least {} data shards, but found only {} healthy disks. (Missing/Corrupt: {}, Parity: {})",
+                                bucket,
+                                object,
+                                required_data,
+                                healthy_count,
+                                disks_to_heal_count,
+                                latest_meta.erasure.parity_blocks
                             );
 
                             // Allow for dangling deletes, on versions that have DataDir missing etc.
@@ -2701,7 +2895,7 @@ impl SetDisks {
                                     Ok((self.default_heal_result(m, &t_errs, bucket, object, version_id).await, Some(derr)))
                                 }
                                 Err(err) => {
-                                    // t_errs = vec![Some(err.clone()); errs.len()];
+                                    // t_errs = vec![Some(err.clone()]; errs.len());
                                     let mut t_errs = Vec::with_capacity(errs.len());
                                     for _ in 0..errs.len() {
                                         t_errs.push(Some(err.clone()));
@@ -2856,7 +3050,7 @@ impl SetDisks {
                                 );
                                 for (index, disk) in latest_disks.iter().enumerate() {
                                     if let Some(outdated_disk) = &out_dated_disks[index] {
-                                        info!("Creating writer for index {} (outdated disk)", index);
+                                        info!(disk_index = index, "Creating writer for outdated disk");
                                         let writer = create_bitrot_writer(
                                             is_inline_buffer,
                                             Some(outdated_disk),
@@ -2869,7 +3063,7 @@ impl SetDisks {
                                         .await?;
                                         writers.push(Some(writer));
                                     } else {
-                                        info!("Skipping writer for index {} (not outdated)", index);
+                                        info!(disk_index = index, "Skipping writer (disk not outdated)");
                                         writers.push(None);
                                     }
 
@@ -2879,7 +3073,7 @@ impl SetDisks {
                                     //     //         Box::new(Cursor::new(Vec::new()))
                                     //     //     } else {
                                     //     //         let disk = disk.clone();
-                                    //     //         let part_path = format!("{}/{}/part.{}", tmp_id, dst_data_dir, part.number);
+                                    //     //         let part_path = format!("{}/{}/part.{}", object, src_data_dir, part.number);
                                     //     //         disk.create_file("", RUSTFS_META_TMP_BUCKET, &part_path, 0).await?
                                     //     //     }
                                     //     // };
@@ -2975,6 +3169,12 @@ impl SetDisks {
                             }
                         }
                         // Rename from tmp location to the actual location.
+                        info!(
+                            "Starting rename phase: {} disks to process (bucket={}, object={})",
+                            out_dated_disks.iter().filter(|d| d.is_some()).count(),
+                            bucket,
+                            object
+                        );
                         for (index, outdated_disk) in out_dated_disks.iter().enumerate() {
                             if let Some(disk) = outdated_disk {
                                 // record the index of the updated disks
@@ -2983,8 +3183,8 @@ impl SetDisks {
                                 parts_metadata[index].set_healing();
 
                                 info!(
-                                    "rename temp data, src_volume: {}, src_path: {}, dst_volume: {}, dst_path: {}",
-                                    RUSTFS_META_TMP_BUCKET, tmp_id, bucket, object
+                                    "Renaming healed data for disk {} (endpoint={}): src_volume={}, src_path={}, dst_volume={}, dst_path={}",
+                                    index, self.set_endpoints[index], RUSTFS_META_TMP_BUCKET, tmp_id, bucket, object
                                 );
                                 let rename_result = disk
                                     .rename_data(RUSTFS_META_TMP_BUCKET, &tmp_id, parts_metadata[index].clone(), bucket, object)
@@ -2992,10 +3192,15 @@ impl SetDisks {
 
                                 if let Err(err) = &rename_result {
                                     info!(
-                                        "rename temp data err: {}. Try fallback to direct xl.meta overwrite...",
-                                        err.to_string()
+                                        error = %err,
+                                        disk_index = index,
+                                        endpoint = %self.set_endpoints[index],
+                                        "Rename failed, attempting fallback"
                                     );
 
+                                    // Preserve temp files for safety
+                                    info!(temp_uuid = %tmp_id, "Rename failed, preserving temporary files for safety");
+
                                     let healthy_index = latest_disks.iter().position(|d| d.is_some()).unwrap_or(0);
 
                                     if let Some(healthy_disk) = &latest_disks[healthy_index] {
@@ -3033,7 +3238,10 @@ impl SetDisks {
                                         ));
                                     }
                                 } else {
-                                    info!("remove temp object, volume: {}, path: {}", RUSTFS_META_TMP_BUCKET, tmp_id);
+                                    info!(
+                                        "Successfully renamed healed data for disk {} (endpoint={}), removing temp files from volume={}, path={}",
+                                        index, self.set_endpoints[index], RUSTFS_META_TMP_BUCKET, tmp_id
+                                    );
 
                                     self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id)
                                         .await
@@ -3367,7 +3575,10 @@ impl SetDisks {
             }
             Ok(m)
         } else {
-            error!("delete_if_dang_ling: is_object_dang_ling errs={:?}", errs);
+            error!(
+                "Object {}/{} is corrupted but not dangling (some parts exist). Preserving data for potential manual recovery. Errors: {:?}",
+                bucket, object, errs
+            );
             Err(DiskError::ErasureReadQuorum)
         }
     }
@@ -3492,7 +3703,7 @@ impl ObjectIO for SetDisks {
         opts: &ObjectOptions,
     ) -> Result<GetObjectReader> {
         // Acquire a shared read-lock early to protect read consistency
-        let _read_lock_guard = if !opts.no_lock {
+        let read_lock_guard = if !opts.no_lock {
             Some(
                 self.fast_lock_manager
                     .acquire_read_lock(bucket, object, self.locker_owner.as_str())
@@ -3554,7 +3765,7 @@ impl ObjectIO for SetDisks {
         // Move the read-lock guard into the task so it lives for the duration of the read
         // let _guard_to_hold = _read_lock_guard; // moved into closure below
         tokio::spawn(async move {
-            // let _guard = _guard_to_hold; // keep guard alive until task ends
+            let _guard = read_lock_guard; // keep guard alive until task ends
             let mut writer = wd;
             if let Err(e) = Self::get_object_with_fileinfo(
                 &bucket,
@@ -3581,7 +3792,8 @@ impl ObjectIO for SetDisks {
 
     #[tracing::instrument(level = "debug", skip(self, data,))]
     async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
-        let disks = self.disks.read().await;
+        let disks_snapshot = self.get_disks_internal().await;
+        let (disks, filtered_online) = self.filter_online_disks(disks_snapshot).await;
 
         // Acquire per-object exclusive lock via RAII guard. It auto-releases asynchronously on drop.
         let _object_lock_guard = if !opts.no_lock {
@@ -3622,6 +3834,14 @@ impl ObjectIO for SetDisks {
             write_quorum += 1
         }
 
+        if filtered_online < write_quorum {
+            warn!(
+                "online disk snapshot {} below write quorum {} for {}/{}; returning erasure write quorum error",
+                filtered_online, write_quorum, bucket, object
+            );
+            return Err(to_object_err(Error::ErasureWriteQuorum, vec![bucket, object]));
+        }
+
         let mut fi = FileInfo::new([bucket, object].join("/").as_str(), data_drives, parity_drives);
 
         fi.version_id = {
@@ -4061,7 +4281,7 @@ impl StorageAPI for SetDisks {
 
         // Acquire locks in batch mode (best effort, matching previous behavior)
         let mut batch = rustfs_lock::BatchLockRequest::new(self.locker_owner.as_str()).with_all_or_nothing(false);
-        let mut unique_objects: std::collections::HashSet<String> = std::collections::HashSet::new();
+        let mut unique_objects: HashSet<String> = HashSet::new();
         for dobj in &objects {
             if unique_objects.insert(dobj.object_name.clone()) {
                 batch = batch.add_write_lock(bucket, dobj.object_name.clone());
@@ -4076,7 +4296,7 @@ impl StorageAPI for SetDisks {
             .collect();
         let _lock_guards = batch_result.guards;
 
-        let failed_map: HashMap<(String, String), rustfs_lock::fast_lock::LockResult> = batch_result
+        let failed_map: HashMap<(String, String), LockResult> = batch_result
             .failed_locks
             .into_iter()
             .map(|(key, err)| ((key.bucket.as_ref().to_string(), key.object.as_ref().to_string()), err))
@@ -4164,7 +4384,6 @@ impl StorageAPI for SetDisks {
 
         for (_, mut fi_vers) in vers_map {
             fi_vers.versions.sort_by(|a, b| a.deleted.cmp(&b.deleted));
-            fi_vers.versions.reverse();
 
             if let Some(index) = fi_vers.versions.iter().position(|fi| fi.deleted) {
                 fi_vers.versions.truncate(index + 1);
@@ -4401,6 +4620,7 @@ impl StorageAPI for SetDisks {
         _max_keys: i32,
         _fetch_owner: bool,
         _start_after: Option<String>,
+        _incl_deleted: bool,
     ) -> Result<ListObjectsV2Info> {
         unimplemented!()
     }
@@ -4423,7 +4643,7 @@ impl StorageAPI for SetDisks {
         _rx: CancellationToken,
         _bucket: &str,
         _prefix: &str,
-        _result: tokio::sync::mpsc::Sender<ObjectInfoOrErr>,
+        _result: Sender<ObjectInfoOrErr>,
         _opts: WalkOptions,
     ) -> Result<()> {
         unimplemented!()
@@ -4455,15 +4675,25 @@ impl StorageAPI for SetDisks {
 
     #[tracing::instrument(skip(self))]
     async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> {
-        let _ = rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
-            bucket.to_string(),
-            Some(object.to_string()),
-            false,
-            Some(HealChannelPriority::Normal),
-            Some(self.pool_index),
-            Some(self.set_index),
-        ))
-        .await;
+        if let Err(e) =
+            rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
+                bucket.to_string(),
+                Some(object.to_string()),
+                false,
+                Some(HealChannelPriority::Normal),
+                Some(self.pool_index),
+                Some(self.set_index),
+            ))
+            .await
+        {
+            warn!(
+                bucket,
+                object,
+                version_id,
+                error = %e,
+                "Failed to enqueue heal request for partial object"
+            );
+        }
         Ok(())
     }
 
@@ -4569,7 +4799,7 @@ impl StorageAPI for SetDisks {
         let tgt_client = match tier_config_mgr.get_driver(&opts.transition.tier).await {
             Ok(client) => client,
             Err(err) => {
-                return Err(Error::other(format!("remote tier error: {}", err)));
+                return Err(Error::other(format!("remote tier error: {err}")));
             }
         };
 
@@ -4749,11 +4979,11 @@ impl StorageAPI for SetDisks {
                 false,
             )?;
             let mut p_reader = PutObjReader::new(hash_reader);
-            if let Err(err) = self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await {
-                return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await;
+            return if let Err(err) = self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await {
+                set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await
             } else {
-                return Ok(());
-            }
+                Ok(())
+            };
         }
 
         let res = self_.clone().new_multipart_upload(bucket, object, &ropts).await?;
@@ -4901,7 +5131,16 @@ impl StorageAPI for SetDisks {
             return Err(Error::other(format!("checksum mismatch: {checksum}")));
         }
 
-        let disks = self.disks.read().await.clone();
+        let disks_snapshot = self.get_disks_internal().await;
+        let (disks, filtered_online) = self.filter_online_disks(disks_snapshot).await;
+
+        if filtered_online < write_quorum {
+            warn!(
+                "online disk snapshot {} below write quorum {} for multipart {}/{}; returning erasure write quorum error",
+                filtered_online, write_quorum, bucket, object
+            );
+            return Err(to_object_err(Error::ErasureWriteQuorum, vec![bucket, object]));
+        }
 
         let shuffle_disks = Self::shuffle_disks(&disks, &fi.erasure.distribution);
 
@@ -5463,6 +5702,17 @@ impl StorageAPI for SetDisks {
         uploaded_parts: Vec<CompletePart>,
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
+        let _object_lock_guard = if !opts.no_lock {
+            Some(
+                self.fast_lock_manager
+                    .acquire_write_lock(bucket, object, self.locker_owner.as_str())
+                    .await
+                    .map_err(|e| Error::other(self.format_lock_error(bucket, object, "write", &e)))?,
+            )
+        } else {
+            None
+        };
+
         let (mut fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
@@ -5581,7 +5831,7 @@ impl StorageAPI for SetDisks {
             }
 
             let ext_part = &curr_fi.parts[i];
-            tracing::info!(target:"rustfs_ecstore::set_disk", part_number = p.part_num, part_size = ext_part.size, part_actual_size = ext_part.actual_size, "Completing multipart part");
+            info!(target:"rustfs_ecstore::set_disk", part_number = p.part_num, part_size = ext_part.size, part_actual_size = ext_part.actual_size, "Completing multipart part");
 
             // Normalize ETags by removing quotes before comparison (PR #592 compatibility)
             let client_etag = p.etag.as_ref().map(|e| rustfs_utils::path::trim_etag(e));
@@ -5837,7 +6087,7 @@ impl StorageAPI for SetDisks {
                     bucket.to_string(),
                     Some(object.to_string()),
                     false,
-                    Some(rustfs_common::heal_channel::HealChannelPriority::Normal),
+                    Some(HealChannelPriority::Normal),
                     Some(self.pool_index),
                     Some(self.set_index),
                 ))
@@ -5897,6 +6147,55 @@ impl StorageAPI for SetDisks {
         version_id: &str,
         opts: &HealOpts,
     ) -> Result<(HealResultItem, Option<Error>)> {
+        let mut effective_object = object.to_string();
+
+        // Optimization: Only attempt correction if the name looks suspicious (quotes or URL encoded)
+        // and the original object does NOT exist.
+        let has_quotes = (effective_object.starts_with('\'') && effective_object.ends_with('\''))
+            || (effective_object.starts_with('"') && effective_object.ends_with('"'));
+        let has_percent = effective_object.contains('%');
+
+        if has_quotes || has_percent {
+            let disks = self.disks.read().await;
+            // 1. Check if the original object exists (lightweight check)
+            let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, &effective_object, version_id, false, false).await?;
+
+            if DiskError::is_all_not_found(&errs) {
+                // Original not found. Try candidates.
+                let mut candidates = Vec::new();
+
+                // Candidate 1: URL Decoded (Priority for web access issues)
+                if has_percent {
+                    if let Ok(decoded) = urlencoding::decode(&effective_object) {
+                        if decoded != effective_object {
+                            candidates.push(decoded.to_string());
+                        }
+                    }
+                }
+
+                // Candidate 2: Quote Stripped (For shell copy-paste issues)
+                if has_quotes && effective_object.len() >= 2 {
+                    candidates.push(effective_object[1..effective_object.len() - 1].to_string());
+                }
+
+                // Check candidates
+                for candidate in candidates {
+                    let (_, errs_cand) =
+                        Self::read_all_fileinfo(&disks, "", bucket, &candidate, version_id, false, false).await?;
+
+                    if !DiskError::is_all_not_found(&errs_cand) {
+                        info!(
+                            "Heal request for object '{}' failed (not found). Auto-corrected to '{}'.",
+                            effective_object, candidate
+                        );
+                        effective_object = candidate;
+                        break; // Found a match, stop searching
+                    }
+                }
+            }
+        }
+        let object = effective_object.as_str();
+
         let _write_lock_guard = if !opts.no_lock {
             let key = rustfs_lock::fast_lock::types::ObjectKey::new(bucket, object);
             let mut skip_lock = false;
@@ -5911,10 +6210,10 @@ impl StorageAPI for SetDisks {
                     skip_lock = true;
                 }
             }
-
             if skip_lock {
                 None
             } else {
+                info!(?opts, "Starting heal_object");
                 Some(
                     self.fast_lock_manager
                         .acquire_write_lock(bucket, object, self.locker_owner.as_str())
@@ -6480,9 +6779,11 @@ fn get_complete_multipart_md5(parts: &[CompletePart]) -> String {
     }
 
     let mut hasher = Md5::new();
-    let _ = hasher.write(&buf);
+    hasher.update(&buf);
 
-    format!("{:x}-{}", hasher.finalize(), parts.len())
+    let digest = hasher.finalize();
+    let etag_hex = faster_hex::hex_string(digest.as_slice());
+    format!("{}-{}", etag_hex, parts.len())
 }
 
 pub fn canonicalize_etag(etag: &str) -> String {
@@ -6562,6 +6863,26 @@ mod tests {
     use std::collections::HashMap;
     use time::OffsetDateTime;
 
+    #[test]
+    fn disk_health_entry_returns_cached_value_within_ttl() {
+        let entry = DiskHealthEntry {
+            last_check: Instant::now(),
+            online: true,
+        };
+
+        assert_eq!(entry.cached_value(), Some(true));
+    }
+
+    #[test]
+    fn disk_health_entry_expires_after_ttl() {
+        let entry = DiskHealthEntry {
+            last_check: Instant::now() - (DISK_HEALTH_CACHE_TTL + Duration::from_millis(100)),
+            online: true,
+        };
+
+        assert!(entry.cached_value().is_none());
+    }
+
     #[test]
     fn test_check_part_constants() {
         // Test that all CHECK_PART constants have expected values

crates/ecstore/src/sets.rs

@@ -111,6 +111,9 @@ impl Sets {
 
         let mut disk_set = Vec::with_capacity(set_count);
 
+        // Create fast lock manager for high performance
+        let fast_lock_manager = Arc::new(rustfs_lock::FastObjectLockManager::new());
+
         for i in 0..set_count {
             let mut set_drive = Vec::with_capacity(set_drive_count);
             let mut set_endpoints = Vec::with_capacity(set_drive_count);
@@ -164,11 +167,9 @@ impl Sets {
 
             // Note: write_quorum was used for the old lock system, no longer needed with FastLock
             let _write_quorum = set_drive_count - parity_count;
-            // Create fast lock manager for high performance
-            let fast_lock_manager = Arc::new(rustfs_lock::FastObjectLockManager::new());
 
             let set_disks = SetDisks::new(
-                fast_lock_manager,
+                fast_lock_manager.clone(),
                 GLOBAL_Local_Node_Name.read().await.to_string(),
                 Arc::new(RwLock::new(set_drive)),
                 set_drive_count,
@@ -439,6 +440,7 @@ impl StorageAPI for Sets {
         _max_keys: i32,
         _fetch_owner: bool,
         _start_after: Option<String>,
+        _incl_deleted: bool,
     ) -> Result<ListObjectsV2Info> {
         unimplemented!()
     }

crates/ecstore/src/store.rs

@@ -1338,9 +1338,19 @@ impl StorageAPI for ECStore {
         max_keys: i32,
         fetch_owner: bool,
         start_after: Option<String>,
+        incl_deleted: bool,
     ) -> Result<ListObjectsV2Info> {
-        self.inner_list_objects_v2(bucket, prefix, continuation_token, delimiter, max_keys, fetch_owner, start_after)
-            .await
+        self.inner_list_objects_v2(
+            bucket,
+            prefix,
+            continuation_token,
+            delimiter,
+            max_keys,
+            fetch_owner,
+            start_after,
+            incl_deleted,
+        )
+        .await
     }
 
     #[instrument(skip(self))]

crates/ecstore/src/store_api.rs

@@ -1224,6 +1224,7 @@ pub trait StorageAPI: ObjectIO + Debug {
         max_keys: i32,
         fetch_owner: bool,
         start_after: Option<String>,
+        incl_deleted: bool,
     ) -> Result<ListObjectsV2Info>;
     // ListObjectVersions TODO: FIXME:
     async fn list_object_versions(

crates/ecstore/src/store_list_objects.rs

@@ -225,6 +225,7 @@ impl ECStore {
         max_keys: i32,
         _fetch_owner: bool,
         start_after: Option<String>,
+        incl_deleted: bool,
     ) -> Result<ListObjectsV2Info> {
         let marker = {
             if continuation_token.is_none() {
@@ -234,7 +235,9 @@ impl ECStore {
             }
         };
 
-        let loi = self.list_objects_generic(bucket, prefix, marker, delimiter, max_keys).await?;
+        let loi = self
+            .list_objects_generic(bucket, prefix, marker, delimiter, max_keys, incl_deleted)
+            .await?;
         Ok(ListObjectsV2Info {
             is_truncated: loi.is_truncated,
             continuation_token,
@@ -251,6 +254,7 @@ impl ECStore {
         marker: Option<String>,
         delimiter: Option<String>,
         max_keys: i32,
+        incl_deleted: bool,
     ) -> Result<ListObjectsInfo> {
         let opts = ListPathOptions {
             bucket: bucket.to_owned(),
@@ -258,7 +262,7 @@ impl ECStore {
             separator: delimiter.clone(),
             limit: max_keys_plus_one(max_keys, marker.is_some()),
             marker,
-            incl_deleted: false,
+            incl_deleted,
             ask_disks: "strict".to_owned(), //TODO: from config
             ..Default::default()
         };

crates/filemeta/Cargo.toml

@@ -26,7 +26,7 @@ categories = ["web-programming", "development-tools", "filesystem"]
 documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
 
 [dependencies]
-crc32fast = { workspace = true }
+crc-fast = { workspace = true }
 rmp.workspace = true
 rmp-serde.workspace = true
 serde.workspace = true

crates/filemeta/src/fileinfo.rs

@@ -220,7 +220,11 @@ impl FileInfo {
         let indices = {
             let cardinality = data_blocks + parity_blocks;
             let mut nums = vec![0; cardinality];
-            let key_crc = crc32fast::hash(object.as_bytes());
+            let key_crc = {
+                let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+                hasher.update(object.as_bytes());
+                hasher.finalize() as u32
+            };
 
             let start = key_crc as usize % cardinality;
             for i in 1..=cardinality {

crates/filemeta/src/filemeta.rs

@@ -427,11 +427,19 @@ impl FileMeta {
             return;
         }
 
-        self.versions.reverse();
-
-        // for _v in self.versions.iter() {
-        //     //  warn!("sort {} {:?}", i, v);
-        // }
+        self.versions.sort_by(|a, b| {
+            if a.header.mod_time != b.header.mod_time {
+                b.header.mod_time.cmp(&a.header.mod_time)
+            } else if a.header.version_type != b.header.version_type {
+                b.header.version_type.cmp(&a.header.version_type)
+            } else if a.header.version_id != b.header.version_id {
+                b.header.version_id.cmp(&a.header.version_id)
+            } else if a.header.flags != b.header.flags {
+                b.header.flags.cmp(&a.header.flags)
+            } else {
+                b.cmp(a)
+            }
+        });
     }
 
     // Find version
@@ -489,25 +497,27 @@ impl FileMeta {
 
         self.versions.sort_by(|a, b| {
             if a.header.mod_time != b.header.mod_time {
-                a.header.mod_time.cmp(&b.header.mod_time)
+                b.header.mod_time.cmp(&a.header.mod_time)
             } else if a.header.version_type != b.header.version_type {
-                a.header.version_type.cmp(&b.header.version_type)
+                b.header.version_type.cmp(&a.header.version_type)
             } else if a.header.version_id != b.header.version_id {
-                a.header.version_id.cmp(&b.header.version_id)
+                b.header.version_id.cmp(&a.header.version_id)
             } else if a.header.flags != b.header.flags {
-                a.header.flags.cmp(&b.header.flags)
+                b.header.flags.cmp(&a.header.flags)
             } else {
-                a.cmp(b)
+                b.cmp(a)
             }
         });
         Ok(())
     }
 
-    pub fn add_version(&mut self, fi: FileInfo) -> Result<()> {
-        let vid = fi.version_id;
+    pub fn add_version(&mut self, mut fi: FileInfo) -> Result<()> {
+        if fi.version_id.is_none() {
+            fi.version_id = Some(Uuid::nil());
+        }
 
         if let Some(ref data) = fi.data {
-            let key = vid.unwrap_or_default().to_string();
+            let key = fi.version_id.unwrap_or_default().to_string();
             self.data.replace(&key, data.to_vec())?;
         }
 
@@ -521,12 +531,13 @@ impl FileMeta {
             return Err(Error::other("file meta version invalid"));
         }
 
-        // 1000 is the limit of versions TODO: make it configurable
-        if self.versions.len() + 1 > 1000 {
-            return Err(Error::other(
-                "You've exceeded the limit on the number of versions you can create on this object",
-            ));
-        }
+        // TODO: make it configurable
+        // 1000 is the limit of versions
+        // if self.versions.len() + 1 > 1000 {
+        //     return Err(Error::other(
+        //         "You've exceeded the limit on the number of versions you can create on this object",
+        //     ));
+        // }
 
         if self.versions.is_empty() {
             self.versions.push(FileMetaShallowVersion::try_from(version)?);
@@ -551,7 +562,6 @@ impl FileMeta {
                 }
             }
         }
-
         Err(Error::other("add_version failed"))
 
         // if !ver.valid() {
@@ -583,12 +593,19 @@ impl FileMeta {
     }
 
     // delete_version deletes version, returns data_dir
+    #[tracing::instrument(skip(self))]
     pub fn delete_version(&mut self, fi: &FileInfo) -> Result<Option<Uuid>> {
+        let vid = if fi.version_id.is_none() {
+            Some(Uuid::nil())
+        } else {
+            Some(fi.version_id.unwrap())
+        };
+
         let mut ventry = FileMetaVersion::default();
         if fi.deleted {
             ventry.version_type = VersionType::Delete;
             ventry.delete_marker = Some(MetaDeleteMarker {
-                version_id: fi.version_id,
+                version_id: vid,
                 mod_time: fi.mod_time,
                 ..Default::default()
             });
@@ -689,8 +706,10 @@ impl FileMeta {
             }
         }
 
+        let mut found_index = None;
+
         for (i, ver) in self.versions.iter().enumerate() {
-            if ver.header.version_id != fi.version_id {
+            if ver.header.version_id != vid {
                 continue;
             }
 
@@ -701,7 +720,7 @@ impl FileMeta {
                         let mut v = self.get_idx(i)?;
                         if v.delete_marker.is_none() {
                             v.delete_marker = Some(MetaDeleteMarker {
-                                version_id: fi.version_id,
+                                version_id: vid,
                                 mod_time: fi.mod_time,
                                 meta_sys: HashMap::new(),
                             });
@@ -767,7 +786,7 @@ impl FileMeta {
                     self.versions.remove(i);
 
                     if (fi.mark_deleted && fi.version_purge_status() != VersionPurgeStatusType::Complete)
-                        || (fi.deleted && fi.version_id.is_none())
+                        || (fi.deleted && vid == Some(Uuid::nil()))
                     {
                         self.add_version_filemata(ventry)?;
                     }
@@ -803,18 +822,11 @@ impl FileMeta {
 
                         return Ok(old_dir);
                     }
+                    found_index = Some(i);
                 }
             }
         }
 
-        let mut found_index = None;
-        for (i, version) in self.versions.iter().enumerate() {
-            if version.header.version_type == VersionType::Object && version.header.version_id == fi.version_id {
-                found_index = Some(i);
-                break;
-            }
-        }
-
         let Some(i) = found_index else {
             if fi.deleted {
                 self.add_version_filemata(ventry)?;
@@ -1521,7 +1533,8 @@ impl FileMetaVersionHeader {
         cur.read_exact(&mut buf)?;
         self.version_id = {
             let id = Uuid::from_bytes(buf);
-            if id.is_nil() { None } else { Some(id) }
+            // if id.is_nil() { None } else { Some(id) }
+            Some(id)
         };
 
         // mod_time
@@ -1695,7 +1708,7 @@ impl MetaObject {
     }
 
     pub fn into_fileinfo(&self, volume: &str, path: &str, all_parts: bool) -> FileInfo {
-        let version_id = self.version_id.filter(|&vid| !vid.is_nil());
+        // let version_id = self.version_id.filter(|&vid| !vid.is_nil());
 
         let parts = if all_parts {
             let mut parts = vec![ObjectPartInfo::default(); self.part_numbers.len()];
@@ -1799,7 +1812,7 @@ impl MetaObject {
             .unwrap_or_default();
 
         FileInfo {
-            version_id,
+            version_id: self.version_id,
             erasure,
             data_dir: self.data_dir,
             mod_time: self.mod_time,

crates/kms/Cargo.toml

@@ -37,7 +37,6 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 tracing = { workspace = true }
 thiserror = { workspace = true }
-once_cell = { workspace = true }
 
 # Cryptography
 aes-gcm = { workspace = true }

crates/kms/src/api_types.rs

@@ -14,7 +14,7 @@
 
 //! API types for KMS dynamic configuration
 
-use crate::config::{KmsBackend, KmsConfig, VaultAuthMethod};
+use crate::config::{BackendConfig, CacheConfig, KmsBackend, KmsConfig, LocalConfig, TlsConfig, VaultAuthMethod, VaultConfig};
 use crate::service_manager::KmsServiceStatus;
 use crate::types::{KeyMetadata, KeyUsage};
 use serde::{Deserialize, Serialize};
@@ -212,12 +212,12 @@ impl From<&KmsConfig> for KmsConfigSummary {
         };
 
         let backend_summary = match &config.backend_config {
-            crate::config::BackendConfig::Local(local_config) => BackendSummary::Local {
+            BackendConfig::Local(local_config) => BackendSummary::Local {
                 key_dir: local_config.key_dir.clone(),
                 has_master_key: local_config.master_key.is_some(),
                 file_permissions: local_config.file_permissions,
             },
-            crate::config::BackendConfig::Vault(vault_config) => BackendSummary::Vault {
+            BackendConfig::Vault(vault_config) => BackendSummary::Vault {
                 address: vault_config.address.clone(),
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
@@ -248,7 +248,7 @@ impl ConfigureLocalKmsRequest {
         KmsConfig {
             backend: KmsBackend::Local,
             default_key_id: self.default_key_id.clone(),
-            backend_config: crate::config::BackendConfig::Local(crate::config::LocalConfig {
+            backend_config: BackendConfig::Local(LocalConfig {
                 key_dir: self.key_dir.clone(),
                 master_key: self.master_key.clone(),
                 file_permissions: self.file_permissions,
@@ -256,7 +256,7 @@ impl ConfigureLocalKmsRequest {
             timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
-            cache_config: crate::config::CacheConfig {
+            cache_config: CacheConfig {
                 max_keys: self.max_cached_keys.unwrap_or(1000),
                 ttl: Duration::from_secs(self.cache_ttl_seconds.unwrap_or(3600)),
                 enable_metrics: true,
@@ -271,7 +271,7 @@ impl ConfigureVaultKmsRequest {
         KmsConfig {
             backend: KmsBackend::Vault,
             default_key_id: self.default_key_id.clone(),
-            backend_config: crate::config::BackendConfig::Vault(crate::config::VaultConfig {
+            backend_config: BackendConfig::Vault(VaultConfig {
                 address: self.address.clone(),
                 auth_method: self.auth_method.clone(),
                 namespace: self.namespace.clone(),
@@ -279,7 +279,7 @@ impl ConfigureVaultKmsRequest {
                 kv_mount: self.kv_mount.clone().unwrap_or_else(|| "secret".to_string()),
                 key_path_prefix: self.key_path_prefix.clone().unwrap_or_else(|| "rustfs/kms/keys".to_string()),
                 tls: if self.skip_tls_verify.unwrap_or(false) {
-                    Some(crate::config::TlsConfig {
+                    Some(TlsConfig {
                         ca_cert_path: None,
                         client_cert_path: None,
                         client_key_path: None,
@@ -292,7 +292,7 @@ impl ConfigureVaultKmsRequest {
             timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
-            cache_config: crate::config::CacheConfig {
+            cache_config: CacheConfig {
                 max_keys: self.max_cached_keys.unwrap_or(1000),
                 ttl: Duration::from_secs(self.cache_ttl_seconds.unwrap_or(3600)),
                 enable_metrics: true,

crates/kms/src/backends/local.rs

@@ -19,12 +19,12 @@ use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::error::{KmsError, Result};
 use crate::types::*;
-use aes_gcm::aead::rand_core::RngCore;
 use aes_gcm::{
     Aes256Gcm, Key, Nonce,
-    aead::{Aead, AeadCore, KeyInit, OsRng},
+    aead::{Aead, KeyInit},
 };
 use async_trait::async_trait;
+use rand::Rng;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::PathBuf;
@@ -105,8 +105,9 @@ impl LocalKmsClient {
         hasher.update(master_key.as_bytes());
         hasher.update(b"rustfs-kms-local"); // Salt to prevent rainbow tables
         let hash = hasher.finalize();
-
-        Ok(*Key::<Aes256Gcm>::from_slice(&hash))
+        let key = Key::<Aes256Gcm>::try_from(hash.as_slice())
+            .map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        Ok(key)
     }
 
     /// Get the file path for a master key
@@ -117,7 +118,6 @@ impl LocalKmsClient {
     /// Load a master key from disk
     async fn load_master_key(&self, key_id: &str) -> Result<MasterKey> {
         let key_path = self.master_key_path(key_id);
-
         if !key_path.exists() {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -127,9 +127,16 @@ impl LocalKmsClient {
 
         // Decrypt key material if master cipher is available
         let _key_material = if let Some(ref cipher) = self.master_cipher {
-            let nonce = Nonce::from_slice(&stored_key.nonce);
+            if stored_key.nonce.len() != 12 {
+                return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
+            }
+
+            let mut nonce_array = [0u8; 12];
+            nonce_array.copy_from_slice(&stored_key.nonce);
+            let nonce = Nonce::from(nonce_array);
+
             cipher
-                .decrypt(nonce, stored_key.encrypted_key_material.as_ref())
+                .decrypt(&nonce, stored_key.encrypted_key_material.as_ref())
                 .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
         } else {
             stored_key.encrypted_key_material
@@ -155,7 +162,10 @@ impl LocalKmsClient {
 
         // Encrypt key material if master cipher is available
         let (encrypted_key_material, nonce) = if let Some(ref cipher) = self.master_cipher {
-            let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
+            let mut nonce_bytes = [0u8; 12];
+            rand::rng().fill(&mut nonce_bytes[..]);
+            let nonce = Nonce::from(nonce_bytes);
+
             let encrypted = cipher
                 .encrypt(&nonce, key_material)
                 .map_err(|e| KmsError::cryptographic_error("encrypt", e.to_string()))?;
@@ -202,7 +212,7 @@ impl LocalKmsClient {
     /// Generate a random 256-bit key
     fn generate_key_material() -> Vec<u8> {
         let mut key_material = vec![0u8; 32]; // 256 bits
-        OsRng.fill_bytes(&mut key_material);
+        rand::rng().fill(&mut key_material[..]);
         key_material
     }
 
@@ -219,9 +229,14 @@ impl LocalKmsClient {
 
         // Decrypt key material if master cipher is available
         let key_material = if let Some(ref cipher) = self.master_cipher {
-            let nonce = Nonce::from_slice(&stored_key.nonce);
+            if stored_key.nonce.len() != 12 {
+                return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
+            }
+            let mut nonce_array = [0u8; 12];
+            nonce_array.copy_from_slice(&stored_key.nonce);
+            let nonce = Nonce::from(nonce_array);
             cipher
-                .decrypt(nonce, stored_key.encrypted_key_material.as_ref())
+                .decrypt(&nonce, stored_key.encrypted_key_material.as_ref())
                 .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
         } else {
             stored_key.encrypted_key_material
@@ -234,25 +249,39 @@ impl LocalKmsClient {
     async fn encrypt_with_master_key(&self, key_id: &str, plaintext: &[u8]) -> Result<(Vec<u8>, Vec<u8>)> {
         // Load the actual master key material
         let key_material = self.get_key_material(key_id).await?;
-        let cipher = Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(&key_material));
+        let key = Key::<Aes256Gcm>::try_from(key_material.as_slice())
+            .map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        let cipher = Aes256Gcm::new(&key);
+
+        let mut nonce_bytes = [0u8; 12];
+        rand::rng().fill(&mut nonce_bytes[..]);
+
+        let nonce = Nonce::from(nonce_bytes);
 
-        let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
         let ciphertext = cipher
             .encrypt(&nonce, plaintext)
             .map_err(|e| KmsError::cryptographic_error("encrypt", e.to_string()))?;
 
-        Ok((ciphertext, nonce.to_vec()))
+        Ok((ciphertext, nonce_bytes.to_vec()))
     }
 
     /// Decrypt data using a master key
     async fn decrypt_with_master_key(&self, key_id: &str, ciphertext: &[u8], nonce: &[u8]) -> Result<Vec<u8>> {
+        if nonce.len() != 12 {
+            return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
+        }
         // Load the actual master key material
         let key_material = self.get_key_material(key_id).await?;
-        let cipher = Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(&key_material));
+        let key = Key::<Aes256Gcm>::try_from(key_material.as_slice())
+            .map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        let cipher = Aes256Gcm::new(&key);
+
+        let mut nonce_array = [0u8; 12];
+        nonce_array.copy_from_slice(nonce);
+        let nonce_ref = Nonce::from(nonce_array);
 
-        let nonce = Nonce::from_slice(nonce);
         let plaintext = cipher
-            .decrypt(nonce, ciphertext)
+            .decrypt(&nonce_ref, ciphertext)
             .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?;
 
         Ok(plaintext)
@@ -275,7 +304,7 @@ impl KmsClient for LocalKmsClient {
         };
 
         let mut plaintext_key = vec![0u8; key_length];
-        OsRng.fill_bytes(&mut plaintext_key);
+        rand::rng().fill(&mut plaintext_key[..]);
 
         // Encrypt the data key with the master key
         let (encrypted_key, nonce) = self.encrypt_with_master_key(&request.master_key_id, &plaintext_key).await?;
@@ -776,9 +805,14 @@ impl KmsBackend for LocalKmsBackend {
 
         // Decrypt the existing key material to preserve it
         let existing_key_material = if let Some(ref cipher) = self.client.master_cipher {
-            let nonce = Nonce::from_slice(&stored_key.nonce);
+            if stored_key.nonce.len() != 12 {
+                return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
+            }
+            let mut nonce_array = [0u8; 12];
+            nonce_array.copy_from_slice(&stored_key.nonce);
+            let nonce = Nonce::from(nonce_array);
             cipher
-                .decrypt(nonce, stored_key.encrypted_key_material.as_ref())
+                .decrypt(&nonce, stored_key.encrypted_key_material.as_ref())
                 .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
         } else {
             stored_key.encrypted_key_material

crates/kms/src/backends/mod.rs

@@ -20,7 +20,6 @@ use async_trait::async_trait;
 use std::collections::HashMap;
 
 pub mod local;
-
 pub mod vault;
 
 /// Abstract KMS client interface that all backends must implement
@@ -201,6 +200,16 @@ pub struct BackendInfo {
 
 impl BackendInfo {
     /// Create a new backend info
+    ///
+    /// # Arguments
+    /// * `backend_type` - The type of the backend
+    /// * `version` - The version of the backend
+    /// * `endpoint` - The endpoint or location of the backend
+    /// * `healthy` - Whether the backend is healthy
+    ///
+    /// # Returns
+    /// A new BackendInfo instance
+    ///
     pub fn new(backend_type: String, version: String, endpoint: String, healthy: bool) -> Self {
         Self {
             backend_type,
@@ -212,6 +221,14 @@ impl BackendInfo {
     }
 
     /// Add metadata to the backend info
+    ///
+    /// # Arguments
+    /// * `key` - Metadata key
+    /// * `value` - Metadata value
+    ///
+    /// # Returns
+    /// Updated BackendInfo instance
+    ///
     pub fn with_metadata(mut self, key: String, value: String) -> Self {
         self.metadata.insert(key, value);
         self

crates/kms/src/cache.rs

@@ -34,6 +34,13 @@ pub struct KmsCache {
 
 impl KmsCache {
     /// Create a new KMS cache with the specified capacity
+    ///
+    /// # Arguments
+    /// * `capacity` - Maximum number of entries in the cache
+    ///
+    /// # Returns
+    /// A new instance of `KmsCache`
+    ///
     pub fn new(capacity: u64) -> Self {
         Self {
             key_metadata_cache: Cache::builder()
@@ -48,22 +55,47 @@ impl KmsCache {
     }
 
     /// Get key metadata from cache
+    ///
+    /// # Arguments
+    /// * `key_id` - The ID of the key to retrieve metadata for
+    ///
+    /// # Returns
+    /// An `Option` containing the `KeyMetadata` if found, or `None` if not found
+    ///
     pub async fn get_key_metadata(&self, key_id: &str) -> Option<KeyMetadata> {
         self.key_metadata_cache.get(key_id).await
     }
 
     /// Put key metadata into cache
+    ///
+    /// # Arguments
+    /// * `key_id` - The ID of the key to store metadata for
+    /// * `metadata` - The `KeyMetadata` to store in the cache
+    ///
     pub async fn put_key_metadata(&mut self, key_id: &str, metadata: &KeyMetadata) {
         self.key_metadata_cache.insert(key_id.to_string(), metadata.clone()).await;
         self.key_metadata_cache.run_pending_tasks().await;
     }
 
     /// Get data key from cache
+    ///
+    /// # Arguments
+    /// * `key_id` - The ID of the key to retrieve the data key for
+    ///
+    /// # Returns
+    /// An `Option` containing the `CachedDataKey` if found, or `None` if not found
+    ///
     pub async fn get_data_key(&self, key_id: &str) -> Option<CachedDataKey> {
         self.data_key_cache.get(key_id).await
     }
 
     /// Put data key into cache
+    ///
+    /// # Arguments
+    /// * `key_id` - The ID of the key to store the data key for
+    /// * `plaintext` - The plaintext data key bytes
+    /// * `ciphertext` - The ciphertext data key bytes
+    ///
     pub async fn put_data_key(&mut self, key_id: &str, plaintext: &[u8], ciphertext: &[u8]) {
         let cached_key = CachedDataKey {
             plaintext: plaintext.to_vec(),
@@ -75,11 +107,19 @@ impl KmsCache {
     }
 
     /// Remove key metadata from cache
+    ///
+    /// # Arguments
+    /// * `key_id` - The ID of the key to remove metadata for
+    ///
     pub async fn remove_key_metadata(&mut self, key_id: &str) {
         self.key_metadata_cache.remove(key_id).await;
     }
 
     /// Remove data key from cache
+    ///
+    /// # Arguments
+    /// * `key_id` - The ID of the key to remove the data key for
+    ///
     pub async fn remove_data_key(&mut self, key_id: &str) {
         self.data_key_cache.remove(key_id).await;
     }
@@ -95,6 +135,10 @@ impl KmsCache {
     }
 
     /// Get cache statistics (hit count, miss count)
+    ///
+    /// # Returns
+    /// A tuple containing total entries and total misses
+    ///
     pub fn stats(&self) -> (u64, u64) {
         let metadata_stats = (
             self.key_metadata_cache.entry_count(),

crates/kms/src/encryption/ciphers.rs

@@ -16,12 +16,12 @@
 
 use crate::error::{KmsError, Result};
 use crate::types::EncryptionAlgorithm;
-use aes_gcm::aead::rand_core::RngCore;
 use aes_gcm::{
     Aes256Gcm, Key, Nonce,
-    aead::{Aead, KeyInit, OsRng},
+    aead::{Aead, KeyInit},
 };
 use chacha20poly1305::ChaCha20Poly1305;
+use rand::Rng;
 
 /// Trait for object encryption ciphers
 #[cfg_attr(not(test), allow(dead_code))]
@@ -52,13 +52,23 @@ pub struct AesCipher {
 
 impl AesCipher {
     /// Create a new AES cipher with the given key
+    ///
+    /// #Arguments
+    /// * `key` - A byte slice representing the AES-256 key (32 bytes)
+    ///
+    /// #Errors
+    /// Returns `KmsError` if the key size is invalid
+    ///
+    /// #Returns
+    /// A Result containing the AesCipher instance
+    ///
     pub fn new(key: &[u8]) -> Result<Self> {
         if key.len() != 32 {
             return Err(KmsError::invalid_key_size(32, key.len()));
         }
 
-        let key = Key::<Aes256Gcm>::from_slice(key);
-        let cipher = Aes256Gcm::new(key);
+        let key = Key::<Aes256Gcm>::try_from(key).map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        let cipher = Aes256Gcm::new(&key);
 
         Ok(Self { cipher })
     }
@@ -70,12 +80,12 @@ impl ObjectCipher for AesCipher {
             return Err(KmsError::invalid_key_size(12, iv.len()));
         }
 
-        let nonce = Nonce::from_slice(iv);
+        let nonce = Nonce::try_from(iv).map_err(|_| KmsError::cryptographic_error("nonce", "Invalid nonce length"))?;
 
         // AES-GCM includes the tag in the ciphertext
         let ciphertext_with_tag = self
             .cipher
-            .encrypt(nonce, aes_gcm::aead::Payload { msg: plaintext, aad })
+            .encrypt(&nonce, aes_gcm::aead::Payload { msg: plaintext, aad })
             .map_err(KmsError::from_aes_gcm_error)?;
 
         // Split ciphertext and tag
@@ -98,7 +108,7 @@ impl ObjectCipher for AesCipher {
             return Err(KmsError::invalid_key_size(self.tag_size(), tag.len()));
         }
 
-        let nonce = Nonce::from_slice(iv);
+        let nonce = Nonce::try_from(iv).map_err(|_| KmsError::cryptographic_error("nonce", "Invalid nonce length"))?;
 
         // Combine ciphertext and tag for AES-GCM
         let mut ciphertext_with_tag = ciphertext.to_vec();
@@ -107,7 +117,7 @@ impl ObjectCipher for AesCipher {
         let plaintext = self
             .cipher
             .decrypt(
-                nonce,
+                &nonce,
                 aes_gcm::aead::Payload {
                     msg: &ciphertext_with_tag,
                     aad,
@@ -142,13 +152,23 @@ pub struct ChaCha20Cipher {
 
 impl ChaCha20Cipher {
     /// Create a new ChaCha20 cipher with the given key
+    ///
+    /// #Arguments
+    /// * `key` - A byte slice representing the ChaCha20-Poly1305 key (32 bytes)
+    ///
+    /// #Errors
+    /// Returns `KmsError` if the key size is invalid
+    ///
+    /// #Returns
+    /// A Result containing the ChaCha20Cipher instance
+    ///
     pub fn new(key: &[u8]) -> Result<Self> {
         if key.len() != 32 {
             return Err(KmsError::invalid_key_size(32, key.len()));
         }
 
-        let key = chacha20poly1305::Key::from_slice(key);
-        let cipher = ChaCha20Poly1305::new(key);
+        let key = chacha20poly1305::Key::try_from(key).map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        let cipher = ChaCha20Poly1305::new(&key);
 
         Ok(Self { cipher })
     }
@@ -160,12 +180,13 @@ impl ObjectCipher for ChaCha20Cipher {
             return Err(KmsError::invalid_key_size(12, iv.len()));
         }
 
-        let nonce = chacha20poly1305::Nonce::from_slice(iv);
+        let nonce =
+            chacha20poly1305::Nonce::try_from(iv).map_err(|_| KmsError::cryptographic_error("nonce", "Invalid nonce length"))?;
 
         // ChaCha20-Poly1305 includes the tag in the ciphertext
         let ciphertext_with_tag = self
             .cipher
-            .encrypt(nonce, chacha20poly1305::aead::Payload { msg: plaintext, aad })
+            .encrypt(&nonce, chacha20poly1305::aead::Payload { msg: plaintext, aad })
             .map_err(KmsError::from_chacha20_error)?;
 
         // Split ciphertext and tag
@@ -188,7 +209,8 @@ impl ObjectCipher for ChaCha20Cipher {
             return Err(KmsError::invalid_key_size(self.tag_size(), tag.len()));
         }
 
-        let nonce = chacha20poly1305::Nonce::from_slice(iv);
+        let nonce =
+            chacha20poly1305::Nonce::try_from(iv).map_err(|_| KmsError::cryptographic_error("nonce", "Invalid nonce length"))?;
 
         // Combine ciphertext and tag for ChaCha20-Poly1305
         let mut ciphertext_with_tag = ciphertext.to_vec();
@@ -197,7 +219,7 @@ impl ObjectCipher for ChaCha20Cipher {
         let plaintext = self
             .cipher
             .decrypt(
-                nonce,
+                &nonce,
                 chacha20poly1305::aead::Payload {
                     msg: &ciphertext_with_tag,
                     aad,
@@ -226,6 +248,14 @@ impl ObjectCipher for ChaCha20Cipher {
 }
 
 /// Create a cipher instance for the given algorithm and key
+///
+/// #Arguments
+/// * `algorithm` - The encryption algorithm to use
+/// * `key` - A byte slice representing the encryption key
+///
+/// #Returns
+/// A Result containing a boxed ObjectCipher instance
+///
 pub fn create_cipher(algorithm: &EncryptionAlgorithm, key: &[u8]) -> Result<Box<dyn ObjectCipher>> {
     match algorithm {
         EncryptionAlgorithm::Aes256 | EncryptionAlgorithm::AwsKms => Ok(Box::new(AesCipher::new(key)?)),
@@ -234,6 +264,13 @@ pub fn create_cipher(algorithm: &EncryptionAlgorithm, key: &[u8]) -> Result<Box<
 }
 
 /// Generate a random IV for the given algorithm
+///
+/// #Arguments
+/// * `algorithm` - The encryption algorithm for which to generate the IV
+///
+/// #Returns
+/// A vector containing the generated IV bytes
+///
 pub fn generate_iv(algorithm: &EncryptionAlgorithm) -> Vec<u8> {
     let iv_size = match algorithm {
         EncryptionAlgorithm::Aes256 | EncryptionAlgorithm::AwsKms => 12,
@@ -241,7 +278,7 @@ pub fn generate_iv(algorithm: &EncryptionAlgorithm) -> Vec<u8> {
     };
 
     let mut iv = vec![0u8; iv_size];
-    OsRng.fill_bytes(&mut iv);
+    rand::rng().fill(&mut iv[..]);
     iv
 }
 

crates/kms/src/encryption/service.rs

@@ -18,6 +18,12 @@ use crate::encryption::ciphers::{create_cipher, generate_iv};
 use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
 use crate::types::*;
+use base64::Engine;
+use rand::random;
+use std::collections::HashMap;
+use std::io::Cursor;
+use tokio::io::{AsyncRead, AsyncReadExt};
+use tracing::{debug, info};
 use zeroize::Zeroize;
 
 /// Data key for object encryption
@@ -36,12 +42,6 @@ impl Drop for DataKey {
         self.plaintext_key.zeroize();
     }
 }
-use base64::Engine;
-use rand::random;
-use std::collections::HashMap;
-use std::io::Cursor;
-use tokio::io::{AsyncRead, AsyncReadExt};
-use tracing::{debug, info};
 
 /// Service for encrypting and decrypting S3 objects with KMS integration
 pub struct ObjectEncryptionService {
@@ -59,51 +59,110 @@ pub struct EncryptionResult {
 
 impl ObjectEncryptionService {
     /// Create a new object encryption service
+    ///
+    /// # Arguments
+    /// * `kms_manager` - KMS manager to use for key operations
+    ///
+    /// # Returns
+    /// New ObjectEncryptionService instance
+    ///
     pub fn new(kms_manager: KmsManager) -> Self {
         Self { kms_manager }
     }
 
     /// Create a new master key (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - CreateKeyRequest with key parameters
+    ///
+    /// # Returns
+    /// CreateKeyResponse with created key details
+    ///
     pub async fn create_key(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
         self.kms_manager.create_key(request).await
     }
 
     /// Describe a master key (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - DescribeKeyRequest with key ID
+    ///
+    /// # Returns
+    /// DescribeKeyResponse with key metadata
+    ///
     pub async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
         self.kms_manager.describe_key(request).await
     }
 
     /// List master keys (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - ListKeysRequest with listing parameters
+    ///
+    /// # Returns
+    /// ListKeysResponse with list of keys
+    ///
     pub async fn list_keys(&self, request: ListKeysRequest) -> Result<ListKeysResponse> {
         self.kms_manager.list_keys(request).await
     }
 
     /// Generate a data encryption key (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - GenerateDataKeyRequest with key parameters
+    ///
+    /// # Returns
+    /// GenerateDataKeyResponse with generated key details
+    ///
     pub async fn generate_data_key(&self, request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
         self.kms_manager.generate_data_key(request).await
     }
 
     /// Get the default key ID
+    ///
+    /// # Returns
+    /// Option with default key ID if configured
+    ///
     pub fn get_default_key_id(&self) -> Option<&String> {
         self.kms_manager.get_default_key_id()
     }
 
     /// Get cache statistics
+    ///
+    /// # Returns
+    /// Option with (hits, misses) if caching is enabled
+    ///
     pub async fn cache_stats(&self) -> Option<(u64, u64)> {
         self.kms_manager.cache_stats().await
     }
 
     /// Clear the cache
+    ///
+    /// # Returns
+    /// Result indicating success or failure
+    ///
     pub async fn clear_cache(&self) -> Result<()> {
         self.kms_manager.clear_cache().await
     }
 
     /// Get backend health status
+    ///
+    /// # Returns
+    /// Result indicating if backend is healthy
+    ///
     pub async fn health_check(&self) -> Result<bool> {
         self.kms_manager.health_check().await
     }
 
     /// Create a data encryption key for object encryption
+    ///
+    /// # Arguments
+    /// * `kms_key_id` - Optional KMS key ID to use (uses default if None)
+    /// * `context` - ObjectEncryptionContext with bucket and object key
+    ///
+    /// # Returns
+    /// Tuple with DataKey and encrypted key blob
+    ///
     pub async fn create_data_key(
         &self,
         kms_key_id: &Option<String>,
@@ -146,6 +205,14 @@ impl ObjectEncryptionService {
     }
 
     /// Decrypt a data encryption key
+    ///
+    /// # Arguments
+    /// * `encrypted_key` - Encrypted data key blob
+    /// * `context` - ObjectEncryptionContext with bucket and object key
+    ///
+    /// # Returns
+    /// DataKey with decrypted key
+    ///
     pub async fn decrypt_data_key(&self, encrypted_key: &[u8], _context: &ObjectEncryptionContext) -> Result<DataKey> {
         let decrypt_request = DecryptRequest {
             ciphertext: encrypted_key.to_vec(),
@@ -429,6 +496,17 @@ impl ObjectEncryptionService {
     }
 
     /// Decrypt object with customer-provided key (SSE-C)
+    ///
+    /// # Arguments
+    /// * `bucket` - S3 bucket name
+    /// * `object_key` - S3 object key
+    /// * `ciphertext` - Encrypted data
+    /// * `metadata` - Encryption metadata
+    /// * `customer_key` - Customer-provided 256-bit key
+    ///
+    /// # Returns
+    /// Decrypted data as a reader
+    ///
     pub async fn decrypt_object_with_customer_key(
         &self,
         bucket: &str,
@@ -481,6 +559,14 @@ impl ObjectEncryptionService {
     }
 
     /// Validate encryption context
+    ///
+    /// # Arguments
+    /// * `actual` - Actual encryption context from metadata
+    /// * `expected` - Expected encryption context to validate against
+    ///
+    /// # Returns
+    /// Result indicating success or context mismatch
+    ///
     fn validate_encryption_context(&self, actual: &HashMap<String, String>, expected: &HashMap<String, String>) -> Result<()> {
         for (key, expected_value) in expected {
             match actual.get(key) {
@@ -499,6 +585,13 @@ impl ObjectEncryptionService {
     }
 
     /// Convert encryption metadata to HTTP headers for S3 compatibility
+    ///
+    /// # Arguments
+    /// * `metadata` - EncryptionMetadata to convert
+    ///
+    /// # Returns
+    /// HashMap of HTTP headers
+    ///
     pub fn metadata_to_headers(&self, metadata: &EncryptionMetadata) -> HashMap<String, String> {
         let mut headers = HashMap::new();
 
@@ -542,6 +635,13 @@ impl ObjectEncryptionService {
     }
 
     /// Parse encryption metadata from HTTP headers
+    ///
+    /// # Arguments
+    /// * `headers` - HashMap of HTTP headers
+    ///
+    /// # Returns
+    /// EncryptionMetadata parsed from headers
+    ///
     pub fn headers_to_metadata(&self, headers: &HashMap<String, String>) -> Result<EncryptionMetadata> {
         let algorithm = headers
             .get("x-amz-server-side-encryption")

crates/kms/src/error.rs

@@ -116,7 +116,7 @@ impl KmsError {
         Self::BackendError { message: message.into() }
     }
 
-    /// Create an access denied error
+    /// Create access denied error
     pub fn access_denied<S: Into<String>>(message: S) -> Self {
         Self::AccessDenied { message: message.into() }
     }
@@ -184,7 +184,7 @@ impl KmsError {
     }
 }
 
-// Convert from standard library errors
+/// Convert from standard library errors
 impl From<std::io::Error> for KmsError {
     fn from(error: std::io::Error) -> Self {
         Self::IoError {
@@ -206,6 +206,13 @@ impl From<serde_json::Error> for KmsError {
 
 impl KmsError {
     /// Create a KMS error from AES-GCM error
+    ///
+    /// #Arguments
+    /// * `error` - The AES-GCM error to convert
+    ///
+    /// #Returns
+    /// * `KmsError` - The corresponding KMS error
+    ///
     pub fn from_aes_gcm_error(error: aes_gcm::Error) -> Self {
         Self::CryptographicError {
             operation: "AES-GCM".to_string(),
@@ -214,6 +221,13 @@ impl KmsError {
     }
 
     /// Create a KMS error from ChaCha20-Poly1305 error
+    ///
+    /// #Arguments
+    /// * `error` - The ChaCha20-Poly1305 error to convert
+    ///
+    /// #Returns
+    /// * `KmsError` - The corresponding KMS error
+    ///
     pub fn from_chacha20_error(error: chacha20poly1305::Error) -> Self {
         Self::CryptographicError {
             operation: "ChaCha20-Poly1305".to_string(),

crates/kms/src/service_manager.rs

@@ -19,7 +19,7 @@ use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::service::ObjectEncryptionService;
 use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
 use tokio::sync::RwLock;
 use tracing::{error, info, warn};
 
@@ -71,7 +71,7 @@ impl KmsServiceManager {
 
     /// Configure KMS with new configuration
     pub async fn configure(&self, new_config: KmsConfig) -> Result<()> {
-        tracing::info!("CLAUDE DEBUG: configure() called with backend: {:?}", new_config.backend);
+        info!("CLAUDE DEBUG: configure() called with backend: {:?}", new_config.backend);
         info!("Configuring KMS with backend: {:?}", new_config.backend);
 
         // Update configuration
@@ -92,7 +92,7 @@ impl KmsServiceManager {
 
     /// Start KMS service with current configuration
     pub async fn start(&self) -> Result<()> {
-        tracing::info!("CLAUDE DEBUG: start() called");
+        info!("CLAUDE DEBUG: start() called");
         let config = {
             let config_guard = self.config.read().await;
             match config_guard.as_ref() {
@@ -254,7 +254,7 @@ impl Default for KmsServiceManager {
 }
 
 /// Global KMS service manager instance
-static GLOBAL_KMS_SERVICE_MANAGER: once_cell::sync::OnceCell<Arc<KmsServiceManager>> = once_cell::sync::OnceCell::new();
+static GLOBAL_KMS_SERVICE_MANAGER: OnceLock<Arc<KmsServiceManager>> = OnceLock::new();
 
 /// Initialize global KMS service manager
 pub fn init_global_kms_service_manager() -> Arc<KmsServiceManager> {
@@ -270,12 +270,12 @@ pub fn get_global_kms_service_manager() -> Option<Arc<KmsServiceManager>> {
 
 /// Get global encryption service (if KMS is running)
 pub async fn get_global_encryption_service() -> Option<Arc<ObjectEncryptionService>> {
-    tracing::info!("CLAUDE DEBUG: get_global_encryption_service called");
+    info!("CLAUDE DEBUG: get_global_encryption_service called");
     let manager = get_global_kms_service_manager().unwrap_or_else(|| {
-        tracing::warn!("CLAUDE DEBUG: KMS service manager not initialized, initializing now as fallback");
+        warn!("CLAUDE DEBUG: KMS service manager not initialized, initializing now as fallback");
         init_global_kms_service_manager()
     });
     let service = manager.get_encryption_service().await;
-    tracing::info!("CLAUDE DEBUG: get_encryption_service returned: {}", service.is_some());
+    info!("CLAUDE DEBUG: get_encryption_service returned: {}", service.is_some());
     service
 }

crates/kms/src/types.rs

@@ -42,6 +42,17 @@ pub struct DataKey {
 
 impl DataKey {
     /// Create a new data key
+    ///
+    /// # Arguments
+    /// * `key_id` - Unique identifier for the key
+    /// * `version` - Key version number
+    /// * `plaintext` - Optional plaintext key material
+    /// * `ciphertext` - Encrypted key material
+    /// * `key_spec` - Key specification (e.g., "AES_256")
+    ///
+    /// # Returns
+    /// A new `DataKey` instance
+    ///
     pub fn new(key_id: String, version: u32, plaintext: Option<Vec<u8>>, ciphertext: Vec<u8>, key_spec: String) -> Self {
         Self {
             key_id,
@@ -55,6 +66,11 @@ impl DataKey {
     }
 
     /// Clear the plaintext key material from memory for security
+    ///
+    /// # Security
+    /// This method zeroes out the plaintext key material before dropping it
+    /// to prevent sensitive data from lingering in memory.
+    ///
     pub fn clear_plaintext(&mut self) {
         if let Some(ref mut plaintext) = self.plaintext {
             // Zero out the memory before dropping
@@ -64,6 +80,14 @@ impl DataKey {
     }
 
     /// Add metadata to the data key
+    ///
+    /// # Arguments
+    /// * `key` - Metadata key
+    /// * `value` - Metadata value
+    ///
+    /// # Returns
+    /// Updated `DataKey` instance with added metadata
+    ///
     pub fn with_metadata(mut self, key: String, value: String) -> Self {
         self.metadata.insert(key, value);
         self
@@ -97,6 +121,15 @@ pub struct MasterKey {
 
 impl MasterKey {
     /// Create a new master key
+    ///
+    /// # Arguments
+    /// * `key_id` - Unique identifier for the key
+    /// * `algorithm` - Key algorithm (e.g., "AES-256")
+    /// * `created_by` - Optional creator/owner of the key
+    ///
+    /// # Returns
+    /// A new `MasterKey` instance
+    ///
     pub fn new(key_id: String, algorithm: String, created_by: Option<String>) -> Self {
         Self {
             key_id,
@@ -113,6 +146,16 @@ impl MasterKey {
     }
 
     /// Create a new master key with description
+    ///
+    /// # Arguments
+    /// * `key_id` - Unique identifier for the key
+    /// * `algorithm` - Key algorithm (e.g., "AES-256")
+    /// * `created_by` - Optional creator/owner of the key
+    /// * `description` - Optional key description
+    ///
+    /// # Returns
+    /// A new `MasterKey` instance with description
+    ///
     pub fn new_with_description(
         key_id: String,
         algorithm: String,
@@ -218,6 +261,14 @@ pub struct GenerateKeyRequest {
 
 impl GenerateKeyRequest {
     /// Create a new generate key request
+    ///
+    /// # Arguments
+    /// * `master_key_id` - Master key ID to use for encryption
+    /// * `key_spec` - Key specification (e.g., "AES_256")
+    ///
+    /// # Returns
+    /// A new `GenerateKeyRequest` instance
+    ///
     pub fn new(master_key_id: String, key_spec: String) -> Self {
         Self {
             master_key_id,
@@ -229,12 +280,27 @@ impl GenerateKeyRequest {
     }
 
     /// Add encryption context
+    ///
+    /// # Arguments
+    /// * `key` - Context key
+    /// * `value` - Context value
+    ///
+    /// # Returns
+    /// Updated `GenerateKeyRequest` instance with added context
+    ///
     pub fn with_context(mut self, key: String, value: String) -> Self {
         self.encryption_context.insert(key, value);
         self
     }
 
     /// Set key length explicitly
+    ///
+    /// # Arguments
+    /// * `length` - Key length in bytes
+    ///
+    /// # Returns
+    /// Updated `GenerateKeyRequest` instance with specified key length
+    ///
     pub fn with_length(mut self, length: u32) -> Self {
         self.key_length = Some(length);
         self
@@ -256,6 +322,14 @@ pub struct EncryptRequest {
 
 impl EncryptRequest {
     /// Create a new encrypt request
+    ///
+    /// # Arguments
+    /// * `key_id` - Key ID to use for encryption
+    /// * `plaintext` - Plaintext data to encrypt
+    ///
+    /// # Returns
+    /// A new `EncryptRequest` instance
+    ///
     pub fn new(key_id: String, plaintext: Vec<u8>) -> Self {
         Self {
             key_id,
@@ -266,6 +340,14 @@ impl EncryptRequest {
     }
 
     /// Add encryption context
+    ///
+    /// # Arguments
+    /// * `key` - Context key
+    /// * `value` - Context value
+    ///
+    /// # Returns
+    /// Updated `EncryptRequest` instance with added context
+    ///
     pub fn with_context(mut self, key: String, value: String) -> Self {
         self.encryption_context.insert(key, value);
         self
@@ -298,6 +380,13 @@ pub struct DecryptRequest {
 
 impl DecryptRequest {
     /// Create a new decrypt request
+    ///
+    /// # Arguments
+    /// * `ciphertext` - Ciphertext to decrypt
+    ///
+    /// # Returns
+    /// A new `DecryptRequest` instance
+    ///
     pub fn new(ciphertext: Vec<u8>) -> Self {
         Self {
             ciphertext,
@@ -307,6 +396,14 @@ impl DecryptRequest {
     }
 
     /// Add encryption context
+    ///
+    /// # Arguments
+    /// * `key` - Context key
+    /// * `value` - Context value
+    ///
+    /// # Returns
+    /// Updated `DecryptRequest` instance with added context
+    ///
     pub fn with_context(mut self, key: String, value: String) -> Self {
         self.encryption_context.insert(key, value);
         self
@@ -365,6 +462,13 @@ pub struct OperationContext {
 
 impl OperationContext {
     /// Create a new operation context
+    ///
+    /// # Arguments
+    /// * `principal` - User or service performing the operation
+    ///
+    /// # Returns
+    /// A new `OperationContext` instance
+    ///
     pub fn new(principal: String) -> Self {
         Self {
             operation_id: Uuid::new_v4(),
@@ -376,18 +480,40 @@ impl OperationContext {
     }
 
     /// Add additional context
+    ///
+    /// # Arguments
+    /// * `key` - Context key
+    /// * `value` - Context value
+    ///
+    /// # Returns
+    /// Updated `OperationContext` instance with added context
+    ///
     pub fn with_context(mut self, key: String, value: String) -> Self {
         self.additional_context.insert(key, value);
         self
     }
 
     /// Set source IP
+    ///
+    /// # Arguments
+    /// * `ip` - Source IP address
+    ///
+    /// # Returns
+    /// Updated `OperationContext` instance with source IP
+    ///
     pub fn with_source_ip(mut self, ip: String) -> Self {
         self.source_ip = Some(ip);
         self
     }
 
     /// Set user agent
+    ///
+    /// # Arguments
+    /// * `agent` - User agent string
+    ///
+    /// # Returns
+    /// Updated `OperationContext` instance with user agent
+    ///
     pub fn with_user_agent(mut self, agent: String) -> Self {
         self.user_agent = Some(agent);
         self
@@ -411,6 +537,14 @@ pub struct ObjectEncryptionContext {
 
 impl ObjectEncryptionContext {
     /// Create a new object encryption context
+    ///
+    /// # Arguments
+    /// * `bucket` - Bucket name
+    /// * `object_key` - Object key
+    ///
+    /// # Returns
+    /// A new `ObjectEncryptionContext` instance
+    ///
     pub fn new(bucket: String, object_key: String) -> Self {
         Self {
             bucket,
@@ -422,18 +556,40 @@ impl ObjectEncryptionContext {
     }
 
     /// Set content type
+    ///
+    /// # Arguments
+    /// * `content_type` - Content type string
+    ///
+    /// # Returns
+    /// Updated `ObjectEncryptionContext` instance with content type
+    ///
     pub fn with_content_type(mut self, content_type: String) -> Self {
         self.content_type = Some(content_type);
         self
     }
 
     /// Set object size
+    ///
+    /// # Arguments
+    /// * `size` - Object size in bytes
+    ///
+    /// # Returns
+    /// Updated `ObjectEncryptionContext` instance with size
+    ///
     pub fn with_size(mut self, size: u64) -> Self {
         self.size = Some(size);
         self
     }
 
     /// Add encryption context
+    ///
+    /// # Arguments
+    /// * `key` - Context key
+    /// * `value` - Context value
+    ///
+    /// # Returns
+    /// Updated `ObjectEncryptionContext` instance with added context
+    ///
     pub fn with_encryption_context(mut self, key: String, value: String) -> Self {
         self.encryption_context.insert(key, value);
         self
@@ -503,6 +659,10 @@ pub enum KeySpec {
 
 impl KeySpec {
     /// Get the key size in bytes
+    ///
+    /// # Returns
+    /// Key size in bytes
+    ///
     pub fn key_size(&self) -> usize {
         match self {
             Self::Aes256 => 32,
@@ -512,6 +672,10 @@ impl KeySpec {
     }
 
     /// Get the string representation for backends
+    ///
+    /// # Returns
+    /// Key specification as a string
+    ///
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Aes256 => "AES_256",
@@ -636,6 +800,14 @@ pub struct GenerateDataKeyRequest {
 
 impl GenerateDataKeyRequest {
     /// Create a new generate data key request
+    ///
+    /// # Arguments
+    /// * `key_id` - Key ID to use for encryption
+    /// * `key_spec` - Key specification
+    ///
+    /// # Returns
+    /// A new `GenerateDataKeyRequest` instance
+    ///
     pub fn new(key_id: String, key_spec: KeySpec) -> Self {
         Self {
             key_id,
@@ -658,6 +830,10 @@ pub struct GenerateDataKeyResponse {
 
 impl EncryptionAlgorithm {
     /// Get the algorithm name as a string
+    ///
+    /// # Returns
+    /// Algorithm name as a string
+    ///
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Aes256 => "AES256",

crates/notify/Cargo.toml

@@ -35,7 +35,6 @@ chrono = { workspace = true, features = ["serde"] }
 futures = { workspace = true }
 form_urlencoded = { workspace = true }
 hashbrown = { workspace = true }
-once_cell = { workspace = true }
 quick-xml = { workspace = true, features = ["serialize", "async-tokio"] }
 rayon = { workspace = true }
 rumqttc = { workspace = true }

crates/notify/src/error.rs

@@ -12,11 +12,22 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-use rustfs_targets::TargetError;
-use rustfs_targets::arn::TargetID;
+use rustfs_targets::{TargetError, arn::TargetID};
 use std::io;
 use thiserror::Error;
 
+/// Errors related to the notification system's lifecycle.
+#[derive(Debug, Error)]
+pub enum LifecycleError {
+    /// Error indicating the system has already been initialized.
+    #[error("System has already been initialized")]
+    AlreadyInitialized,
+
+    /// Error indicating the system has not been initialized yet.
+    #[error("System has not been initialized")]
+    NotInitialized,
+}
+
 /// Error types for the notification system
 #[derive(Debug, Error)]
 pub enum NotificationError {
@@ -38,11 +49,8 @@ pub enum NotificationError {
     #[error("Rule configuration error: {0}")]
     RuleConfiguration(String),
 
-    #[error("System initialization error: {0}")]
-    Initialization(String),
-
-    #[error("Notification system has already been initialized")]
-    AlreadyInitialized,
+    #[error("System lifecycle error: {0}")]
+    Lifecycle(#[from] LifecycleError),
 
     #[error("I/O error: {0}")]
     Io(io::Error),
@@ -56,6 +64,9 @@ pub enum NotificationError {
     #[error("Target '{0}' not found")]
     TargetNotFound(TargetID),
 
-    #[error("Server not initialized")]
-    ServerNotInitialized,
+    #[error("System initialization error: {0}")]
+    Initialization(String),
+
+    #[error("Storage not available: {0}")]
+    StorageNotAvailable(String),
 }

crates/notify/src/event.rs

@@ -276,3 +276,120 @@ impl EventArgs {
         self.req_params.contains_key("x-rustfs-source-replication-request")
     }
 }
+
+/// Builder for [`EventArgs`].
+///
+/// This builder provides a fluent API to construct an `EventArgs` instance,
+/// ensuring that all required fields are provided.
+///
+/// # Example
+///
+/// ```ignore
+/// let args = EventArgsBuilder::new(
+///     EventName::ObjectCreatedPut,
+///     "my-bucket",
+///     object_info,
+/// )
+/// .host("localhost:9000")
+/// .user_agent("my-app/1.0")
+/// .build();
+/// ```
+#[derive(Debug, Clone, Default)]
+pub struct EventArgsBuilder {
+    event_name: EventName,
+    bucket_name: String,
+    object: rustfs_ecstore::store_api::ObjectInfo,
+    req_params: HashMap<String, String>,
+    resp_elements: HashMap<String, String>,
+    version_id: String,
+    host: String,
+    user_agent: String,
+}
+
+impl EventArgsBuilder {
+    /// Creates a new builder with the required fields.
+    pub fn new(event_name: EventName, bucket_name: impl Into<String>, object: rustfs_ecstore::store_api::ObjectInfo) -> Self {
+        Self {
+            event_name,
+            bucket_name: bucket_name.into(),
+            object,
+            ..Default::default()
+        }
+    }
+
+    /// Sets the event name.
+    pub fn event_name(mut self, event_name: EventName) -> Self {
+        self.event_name = event_name;
+        self
+    }
+
+    /// Sets the bucket name.
+    pub fn bucket_name(mut self, bucket_name: impl Into<String>) -> Self {
+        self.bucket_name = bucket_name.into();
+        self
+    }
+
+    /// Sets the object information.
+    pub fn object(mut self, object: rustfs_ecstore::store_api::ObjectInfo) -> Self {
+        self.object = object;
+        self
+    }
+
+    /// Sets the request parameters.
+    pub fn req_params(mut self, req_params: HashMap<String, String>) -> Self {
+        self.req_params = req_params;
+        self
+    }
+
+    /// Adds a single request parameter.
+    pub fn req_param(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
+        self.req_params.insert(key.into(), value.into());
+        self
+    }
+
+    /// Sets the response elements.
+    pub fn resp_elements(mut self, resp_elements: HashMap<String, String>) -> Self {
+        self.resp_elements = resp_elements;
+        self
+    }
+
+    /// Adds a single response element.
+    pub fn resp_element(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
+        self.resp_elements.insert(key.into(), value.into());
+        self
+    }
+
+    /// Sets the version ID.
+    pub fn version_id(mut self, version_id: impl Into<String>) -> Self {
+        self.version_id = version_id.into();
+        self
+    }
+
+    /// Sets the host.
+    pub fn host(mut self, host: impl Into<String>) -> Self {
+        self.host = host.into();
+        self
+    }
+
+    /// Sets the user agent.
+    pub fn user_agent(mut self, user_agent: impl Into<String>) -> Self {
+        self.user_agent = user_agent.into();
+        self
+    }
+
+    /// Builds the final `EventArgs` instance.
+    ///
+    /// This method consumes the builder and returns the constructed `EventArgs`.
+    pub fn build(self) -> EventArgs {
+        EventArgs {
+            event_name: self.event_name,
+            bucket_name: self.bucket_name,
+            object: self.object,
+            req_params: self.req_params,
+            resp_elements: self.resp_elements,
+            version_id: self.version_id,
+            host: self.host,
+            user_agent: self.user_agent,
+        }
+    }
+}

crates/notify/src/global.rs

@@ -12,17 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::{BucketNotificationConfig, Event, EventArgs, NotificationError, NotificationSystem};
-use once_cell::sync::Lazy;
+use crate::{BucketNotificationConfig, Event, EventArgs, LifecycleError, NotificationError, NotificationSystem};
 use rustfs_ecstore::config::Config;
-use rustfs_targets::EventName;
-use rustfs_targets::arn::TargetID;
+use rustfs_targets::{EventName, arn::TargetID};
 use std::sync::{Arc, OnceLock};
-use tracing::{error, instrument};
+use tracing::error;
 
 static NOTIFICATION_SYSTEM: OnceLock<Arc<NotificationSystem>> = OnceLock::new();
-// Create a globally unique Notifier instance
-static GLOBAL_NOTIFIER: Lazy<Notifier> = Lazy::new(|| Notifier {});
 
 /// Initialize the global notification system with the given configuration.
 /// This function should only be called once throughout the application life cycle.
@@ -34,7 +30,7 @@ pub async fn initialize(config: Config) -> Result<(), NotificationError> {
 
     match NOTIFICATION_SYSTEM.set(Arc::new(system)) {
         Ok(_) => Ok(()),
-        Err(_) => Err(NotificationError::AlreadyInitialized),
+        Err(_) => Err(NotificationError::Lifecycle(LifecycleError::AlreadyInitialized)),
     }
 }
 
@@ -49,14 +45,11 @@ pub fn is_notification_system_initialized() -> bool {
     NOTIFICATION_SYSTEM.get().is_some()
 }
 
-/// Returns a reference to the global Notifier instance.
-pub fn notifier_instance() -> &'static Notifier {
-    &GLOBAL_NOTIFIER
-}
+/// A module providing the public API for event notification.
+pub mod notifier_global {
+    use super::*;
+    use tracing::instrument;
 
-pub struct Notifier {}
-
-impl Notifier {
     /// Notify an event asynchronously.
     /// This is the only entry point for all event notifications in the system.
     /// # Parameter
@@ -67,8 +60,8 @@ impl Notifier {
     ///
     /// # Using
     /// This function is used to notify events in the system, such as object creation, deletion, or updates.
-    #[instrument(skip(self, args))]
-    pub async fn notify(&self, args: EventArgs) {
+    #[instrument(skip(args))]
+    pub async fn notify(args: EventArgs) {
         // Dependency injection or service positioning mode obtain NotificationSystem instance
         let notification_sys = match notification_system() {
             // If the notification system itself cannot be retrieved, it will be returned directly
@@ -110,7 +103,6 @@ impl Notifier {
     /// # Using
     /// This function allows you to dynamically add notification rules for a specific bucket.
     pub async fn add_bucket_notification_rule(
-        &self,
         bucket_name: &str,
         region: &str,
         event_names: &[EventName],
@@ -137,7 +129,7 @@ impl Notifier {
         // Get global NotificationSystem
         let notification_sys = match notification_system() {
             Some(sys) => sys,
-            None => return Err(NotificationError::ServerNotInitialized),
+            None => return Err(NotificationError::Lifecycle(LifecycleError::NotInitialized)),
         };
 
         // Loading configuration
@@ -159,7 +151,6 @@ impl Notifier {
     /// # Using
     /// Supports notification rules for adding multiple event types, prefixes, suffixes, and targets to the same bucket in batches.
     pub async fn add_event_specific_rules(
-        &self,
         bucket_name: &str,
         region: &str,
         event_rules: &[(Vec<EventName>, String, String, Vec<TargetID>)],
@@ -176,10 +167,7 @@ impl Notifier {
         }
 
         // Get global NotificationSystem instance
-        let notification_sys = match notification_system() {
-            Some(sys) => sys,
-            None => return Err(NotificationError::ServerNotInitialized),
-        };
+        let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?;
 
         // Loading configuration
         notification_sys
@@ -196,12 +184,9 @@ impl Notifier {
     /// This function allows you to clear all notification rules for a specific bucket.
     /// This is useful when you want to reset the notification configuration for a bucket.
     ///
-    pub async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError> {
+    pub async fn clear_bucket_notification_rules(bucket_name: &str) -> Result<(), NotificationError> {
         // Get global NotificationSystem instance
-        let notification_sys = match notification_system() {
-            Some(sys) => sys,
-            None => return Err(NotificationError::ServerNotInitialized),
-        };
+        let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?;
 
         // Clear configuration
         notification_sys.remove_bucket_notification_config(bucket_name).await;

crates/notify/src/integration.rs

@@ -140,7 +140,7 @@ impl NotificationSystem {
             info!("Initializing target: {}", target.id());
             // Initialize the target
             if let Err(e) = target.init().await {
-                error!("Target {} Initialization failed:{}", target.id(), e);
+                warn!("Target {} Initialization failed:{}", target.id(), e);
                 continue;
             }
             debug!("Target {} initialized successfully,enabled:{}", target_id, target.is_enabled());
@@ -199,7 +199,9 @@ impl NotificationSystem {
         F: FnMut(&mut Config) -> bool, // The closure returns a boolean value indicating whether the configuration has been changed
     {
         let Some(store) = rustfs_ecstore::global::new_object_layer_fn() else {
-            return Err(NotificationError::ServerNotInitialized);
+            return Err(NotificationError::StorageNotAvailable(
+                "Failed to save target configuration: server storage not initialized".to_string(),
+            ));
         };
 
         let mut new_config = rustfs_ecstore::config::com::read_config_without_migrate(store.clone())
@@ -420,7 +422,7 @@ impl NotificationSystem {
             if !e.to_string().contains("ARN not found") {
                 return Err(NotificationError::BucketNotification(e.to_string()));
             } else {
-                error!("{}", e);
+                error!("config validate failed, err: {}", e);
             }
         }
 

crates/notify/src/lib.rs

@@ -18,18 +18,18 @@
 //! It supports sending events to various targets
 //! (like Webhook and MQTT) and includes features like event persistence and retry on failure.
 
-pub mod error;
-pub mod event;
+mod error;
+mod event;
 pub mod factory;
-pub mod global;
+mod global;
 pub mod integration;
 pub mod notifier;
 pub mod registry;
 pub mod rules;
 pub mod stream;
-// Re-exports
-pub use error::NotificationError;
-pub use event::{Event, EventArgs};
-pub use global::{initialize, is_notification_system_initialized, notification_system};
+
+pub use error::{LifecycleError, NotificationError};
+pub use event::{Event, EventArgs, EventArgsBuilder};
+pub use global::{initialize, is_notification_system_initialized, notification_system, notifier_global};
 pub use integration::NotificationSystem;
 pub use rules::BucketNotificationConfig;

crates/obs/Cargo.toml

@@ -38,14 +38,13 @@ rustfs-config = { workspace = true, features = ["constants", "observability"] }
 rustfs-utils = { workspace = true, features = ["ip", "path"] }
 flexi_logger = { workspace = true }
 metrics = { workspace = true }
-metrics-exporter-opentelemetry = { workspace = true }
 nu-ansi-term = { workspace = true }
 nvml-wrapper = { workspace = true, optional = true }
 opentelemetry = { workspace = true }
 opentelemetry-appender-tracing = { workspace = true, features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] }
 opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 opentelemetry-stdout = { workspace = true }
-opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "gzip-tonic", "trace", "metrics", "logs", "internal-logs"] }
+opentelemetry-otlp = { workspace = true }
 opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"] }
 serde = { workspace = true }
 smallvec = { workspace = true, features = ["serde"] }

crates/obs/src/config.rs

@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use rustfs_config::observability::{
-    ENV_OBS_ENDPOINT, ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_KEEP_FILES,
-    ENV_OBS_LOG_ROTATION_SIZE_MB, ENV_OBS_LOG_ROTATION_TIME, ENV_OBS_LOG_STDOUT_ENABLED, ENV_OBS_LOGGER_LEVEL,
-    ENV_OBS_METER_INTERVAL, ENV_OBS_SAMPLE_RATIO, ENV_OBS_SERVICE_NAME, ENV_OBS_SERVICE_VERSION, ENV_OBS_USE_STDOUT,
+    DEFAULT_OBS_ENVIRONMENT_PRODUCTION, ENV_OBS_ENDPOINT, ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_ENDPOINT,
+    ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_KEEP_FILES, ENV_OBS_LOG_ROTATION_SIZE_MB, ENV_OBS_LOG_ROTATION_TIME,
+    ENV_OBS_LOG_STDOUT_ENABLED, ENV_OBS_LOGGER_LEVEL, ENV_OBS_METER_INTERVAL, ENV_OBS_METRIC_ENDPOINT, ENV_OBS_SAMPLE_RATIO,
+    ENV_OBS_SERVICE_NAME, ENV_OBS_SERVICE_VERSION, ENV_OBS_TRACE_ENDPOINT, ENV_OBS_USE_STDOUT,
 };
 use rustfs_config::{
     APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_LOG_ROTATION_SIZE_MB, DEFAULT_LOG_ROTATION_TIME,
@@ -23,6 +24,7 @@ use rustfs_config::{
     USE_STDOUT,
 };
 use rustfs_utils::dirs::get_log_directory_to_string;
+use rustfs_utils::{get_env_bool, get_env_f64, get_env_opt_str, get_env_str, get_env_u64, get_env_usize};
 use serde::{Deserialize, Serialize};
 use std::env;
 
@@ -55,6 +57,9 @@ use std::env;
 #[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct OtelConfig {
     pub endpoint: String,                 // Endpoint for metric collection
+    pub trace_endpoint: Option<String>,   // Endpoint for trace collection
+    pub metric_endpoint: Option<String>,  // Endpoint for metric collection
+    pub log_endpoint: Option<String>,     // Endpoint for log collection
     pub use_stdout: Option<bool>,         // Output to stdout
     pub sample_ratio: Option<f64>,        // Trace sampling ratio
     pub meter_interval: Option<u64>,      // Metric collection interval
@@ -68,7 +73,7 @@ pub struct OtelConfig {
     pub log_filename: Option<String>,      // The name of the log file
     pub log_rotation_size_mb: Option<u64>, // Log file size cut threshold (MB)
     pub log_rotation_time: Option<String>, // Logs are cut by time (Hour， Day，Minute， Second)
-    pub log_keep_files: Option<u16>,       // Number of log files to be retained
+    pub log_keep_files: Option<usize>,     // Number of log files to be retained
 }
 
 impl OtelConfig {
@@ -83,62 +88,29 @@ impl OtelConfig {
         } else {
             env::var(ENV_OBS_ENDPOINT).unwrap_or_else(|_| "".to_string())
         };
-        let mut use_stdout = env::var(ENV_OBS_USE_STDOUT)
-            .ok()
-            .and_then(|v| v.parse().ok())
-            .or(Some(USE_STDOUT));
+        let mut use_stdout = get_env_bool(ENV_OBS_USE_STDOUT, USE_STDOUT);
         if endpoint.is_empty() {
-            use_stdout = Some(true);
+            use_stdout = true;
         }
 
         OtelConfig {
             endpoint,
-            use_stdout,
-            sample_ratio: env::var(ENV_OBS_SAMPLE_RATIO)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(SAMPLE_RATIO)),
-            meter_interval: env::var(ENV_OBS_METER_INTERVAL)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(METER_INTERVAL)),
-            service_name: env::var(ENV_OBS_SERVICE_NAME)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(APP_NAME.to_string())),
-            service_version: env::var(ENV_OBS_SERVICE_VERSION)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(SERVICE_VERSION.to_string())),
-            environment: env::var(ENV_OBS_ENVIRONMENT)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(ENVIRONMENT.to_string())),
-            logger_level: env::var(ENV_OBS_LOGGER_LEVEL)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(DEFAULT_LOG_LEVEL.to_string())),
-            log_stdout_enabled: env::var(ENV_OBS_LOG_STDOUT_ENABLED)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(DEFAULT_OBS_LOG_STDOUT_ENABLED)),
+            trace_endpoint: get_env_opt_str(ENV_OBS_TRACE_ENDPOINT),
+            metric_endpoint: get_env_opt_str(ENV_OBS_METRIC_ENDPOINT),
+            log_endpoint: get_env_opt_str(ENV_OBS_LOG_ENDPOINT),
+            use_stdout: Some(use_stdout),
+            sample_ratio: Some(get_env_f64(ENV_OBS_SAMPLE_RATIO, SAMPLE_RATIO)),
+            meter_interval: Some(get_env_u64(ENV_OBS_METER_INTERVAL, METER_INTERVAL)),
+            service_name: Some(get_env_str(ENV_OBS_SERVICE_NAME, APP_NAME)),
+            service_version: Some(get_env_str(ENV_OBS_SERVICE_VERSION, SERVICE_VERSION)),
+            environment: Some(get_env_str(ENV_OBS_ENVIRONMENT, ENVIRONMENT)),
+            logger_level: Some(get_env_str(ENV_OBS_LOGGER_LEVEL, DEFAULT_LOG_LEVEL)),
+            log_stdout_enabled: Some(get_env_bool(ENV_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOG_STDOUT_ENABLED)),
             log_directory: Some(get_log_directory_to_string(ENV_OBS_LOG_DIRECTORY)),
-            log_filename: env::var(ENV_OBS_LOG_FILENAME)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(DEFAULT_OBS_LOG_FILENAME.to_string())),
-            log_rotation_size_mb: env::var(ENV_OBS_LOG_ROTATION_SIZE_MB)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(DEFAULT_LOG_ROTATION_SIZE_MB)), // Default to 100 MB
-            log_rotation_time: env::var(ENV_OBS_LOG_ROTATION_TIME)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(DEFAULT_LOG_ROTATION_TIME.to_string())), // Default to "Day"
-            log_keep_files: env::var(ENV_OBS_LOG_KEEP_FILES)
-                .ok()
-                .and_then(|v| v.parse().ok())
-                .or(Some(DEFAULT_LOG_KEEP_FILES)), // Default to keeping 30 log files
+            log_filename: Some(get_env_str(ENV_OBS_LOG_FILENAME, DEFAULT_OBS_LOG_FILENAME)),
+            log_rotation_size_mb: Some(get_env_u64(ENV_OBS_LOG_ROTATION_SIZE_MB, DEFAULT_LOG_ROTATION_SIZE_MB)), // Default to 100 MB
+            log_rotation_time: Some(get_env_str(ENV_OBS_LOG_ROTATION_TIME, DEFAULT_LOG_ROTATION_TIME)), // Default to "Hour"
+            log_keep_files: Some(get_env_usize(ENV_OBS_LOG_KEEP_FILES, DEFAULT_LOG_KEEP_FILES)), // Default to keeping 30 log files
         }
     }
 
@@ -237,3 +209,12 @@ impl Default for AppConfig {
         Self::new()
     }
 }
+
+/// Check if the current environment is production
+///
+/// # Returns
+/// true if production, false otherwise
+///
+pub fn is_production_environment() -> bool {
+    get_env_str(ENV_OBS_ENVIRONMENT, ENVIRONMENT).eq_ignore_ascii_case(DEFAULT_OBS_ENVIRONMENT_PRODUCTION)
+}

crates/obs/src/error.rs

@@ -0,0 +1,75 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::OtelGuard;
+use std::sync::{Arc, Mutex};
+use tokio::sync::SetError;
+
+/// Error type for global guard operations
+#[derive(Debug, thiserror::Error)]
+pub enum GlobalError {
+    /// Occurs when attempting to set a global recorder (e.g., via [`crate::Recorder::install_global`] or [`metrics::set_global_recorder`])
+    /// but a global recorder is already initialized.
+    ///
+    /// [`crate::Recorder::install_global`]: crate::Recorder::install_global
+    /// [`metrics::set_global_recorder`]: https://docs.rs/metrics/latest/metrics/fn.set_global_recorder.html
+    #[error("Failed to set a global recorder: {0}")]
+    SetRecorder(#[from] metrics::SetRecorderError<crate::Recorder>),
+    #[error("Failed to set global guard: {0}")]
+    SetError(#[from] SetError<Arc<Mutex<OtelGuard>>>),
+    #[error("Global guard not initialized")]
+    NotInitialized,
+    #[error("Global system metrics err: {0}")]
+    MetricsError(String),
+    #[error("Failed to get current PID: {0}")]
+    PidError(String),
+    #[error("Process with PID {0} not found")]
+    ProcessNotFound(u32),
+    #[error("Failed to get physical core count")]
+    CoreCountError,
+    #[error("GPU initialization failed: {0}")]
+    GpuInitError(String),
+    #[error("GPU device not found: {0}")]
+    GpuDeviceError(String),
+    #[error("Failed to send log: {0}")]
+    SendFailed(&'static str),
+    #[error("Operation timed out: {0}")]
+    Timeout(&'static str),
+    #[error("Telemetry initialization failed: {0}")]
+    TelemetryError(#[from] TelemetryError),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TelemetryError {
+    #[error("Span exporter build failed: {0}")]
+    BuildSpanExporter(String),
+    #[error("Metric exporter build failed: {0}")]
+    BuildMetricExporter(String),
+    #[error("Log exporter build failed: {0}")]
+    BuildLogExporter(String),
+    #[error("Install metrics recorder failed: {0}")]
+    InstallMetricsRecorder(String),
+    #[error("Tracing subscriber init failed: {0}")]
+    SubscriberInit(String),
+    #[error("I/O error: {0}")]
+    Io(String),
+    #[error("Set permissions failed: {0}")]
+    SetPermissions(String),
+}
+
+impl From<std::io::Error> for TelemetryError {
+    fn from(e: std::io::Error) -> Self {
+        TelemetryError::Io(e.to_string())
+    }
+}

crates/obs/src/global.rs

@@ -12,46 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::telemetry::{OtelGuard, init_telemetry};
-use crate::{AppConfig, SystemObserver};
+use crate::{AppConfig, GlobalError, OtelGuard, SystemObserver, telemetry::init_telemetry};
 use std::sync::{Arc, Mutex};
-use tokio::sync::{OnceCell, SetError};
+use tokio::sync::OnceCell;
 use tracing::{error, info};
 
 /// Global guard for OpenTelemetry tracing
 static GLOBAL_GUARD: OnceCell<Arc<Mutex<OtelGuard>>> = OnceCell::const_new();
 
-/// Flag indicating if observability is enabled
-pub(crate) static IS_OBSERVABILITY_ENABLED: OnceCell<bool> = OnceCell::const_new();
+/// Flag indicating if observability metric is enabled
+pub(crate) static OBSERVABILITY_METRIC_ENABLED: OnceCell<bool> = OnceCell::const_new();
 
-/// Check whether Observability is enabled
-pub fn is_observability_enabled() -> bool {
-    IS_OBSERVABILITY_ENABLED.get().copied().unwrap_or(false)
-}
-
-/// Error type for global guard operations
-#[derive(Debug, thiserror::Error)]
-pub enum GlobalError {
-    #[error("Failed to set global guard: {0}")]
-    SetError(#[from] SetError<Arc<Mutex<OtelGuard>>>),
-    #[error("Global guard not initialized")]
-    NotInitialized,
-    #[error("Global system metrics err: {0}")]
-    MetricsError(String),
-    #[error("Failed to get current PID: {0}")]
-    PidError(String),
-    #[error("Process with PID {0} not found")]
-    ProcessNotFound(u32),
-    #[error("Failed to get physical core count")]
-    CoreCountError,
-    #[error("GPU initialization failed: {0}")]
-    GpuInitError(String),
-    #[error("GPU device not found: {0}")]
-    GpuDeviceError(String),
-    #[error("Failed to send log: {0}")]
-    SendFailed(&'static str),
-    #[error("Operation timed out: {0}")]
-    Timeout(&'static str),
+/// Check whether Observability metric is enabled
+pub fn observability_metric_enabled() -> bool {
+    OBSERVABILITY_METRIC_ENABLED.get().copied().unwrap_or(false)
 }
 
 /// Initialize the observability module
@@ -68,14 +42,17 @@ pub enum GlobalError {
 ///
 /// # #[tokio::main]
 /// # async fn main() {
-/// #    let guard = init_obs(None).await;
+/// #    match init_obs(None).await {
+/// #         Ok(guard) => {}
+/// #         Err(e) => { eprintln!("Failed to initialize observability: {}", e); }
+/// #     }
 /// # }
 /// ```
-pub async fn init_obs(endpoint: Option<String>) -> OtelGuard {
+pub async fn init_obs(endpoint: Option<String>) -> Result<OtelGuard, GlobalError> {
     // Load the configuration file
     let config = AppConfig::new_with_endpoint(endpoint);
 
-    let otel_guard = init_telemetry(&config.observability);
+    let otel_guard = init_telemetry(&config.observability)?;
     // Server will be created per connection - this ensures isolation
     tokio::spawn(async move {
         // Record the PID-related metrics of the current process
@@ -90,10 +67,10 @@ pub async fn init_obs(endpoint: Option<String>) -> OtelGuard {
         }
     });
 
-    otel_guard
+    Ok(otel_guard)
 }
 
-/// Set the global guard for OpenTelemetry
+/// Set the global guard for OtelGuard
 ///
 /// # Arguments
 /// * `guard` - The OtelGuard instance to set globally
@@ -107,17 +84,20 @@ pub async fn init_obs(endpoint: Option<String>) -> OtelGuard {
 /// # use rustfs_obs::{ init_obs, set_global_guard};
 ///
 /// # async fn init() -> Result<(), Box<dyn std::error::Error>> {
-/// #    let guard = init_obs(None).await;
+/// #    let guard = match init_obs(None).await{
+/// #         Ok(g) => g,
+/// #         Err(e) => { return Err(Box::new(e)); }
+/// #    };
 /// #    set_global_guard(guard)?;
 /// #    Ok(())
 /// # }
 /// ```
 pub fn set_global_guard(guard: OtelGuard) -> Result<(), GlobalError> {
-    info!("Initializing global OpenTelemetry guard");
+    info!("Initializing global guard");
     GLOBAL_GUARD.set(Arc::new(Mutex::new(guard))).map_err(GlobalError::SetError)
 }
 
-/// Get the global guard for OpenTelemetry
+/// Get the global guard for OtelGuard
 ///
 /// # Returns
 /// * `Ok(Arc<Mutex<OtelGuard>>)` if guard exists

crates/obs/src/lib.rs

@@ -38,15 +38,33 @@
 ///
 /// # #[tokio::main]
 /// # async fn main() {
-/// #   let guard = init_obs(None).await;
+/// #   let _guard = match init_obs(None).await {
+/// #         Ok(g) => g,
+/// #         Err(e) => {
+/// #             panic!("Failed to initialize observability: {:?}", e);
+/// #         }
+/// #     };
+/// #   // Application logic here
+/// #   {
+/// #       // Simulate some work
+/// #       tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+/// #       println!("Application is running...");
+/// #   }
+/// #   // Guard will be dropped here, flushing telemetry data
 /// # }
 /// ```
 mod config;
+mod error;
 mod global;
 mod metrics;
+mod recorder;
 mod system;
 mod telemetry;
 
-pub use config::{AppConfig, OtelConfig};
+pub use config::*;
+pub use error::*;
 pub use global::*;
+pub use metrics::*;
+pub use recorder::*;
 pub use system::SystemObserver;
+pub use telemetry::OtelGuard;

crates/obs/src/metrics/audit.rs

@@ -17,10 +17,15 @@
 /// audit related metric descriptors
 ///
 /// This module contains the metric descriptors for the audit subsystem.
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 const TARGET_ID: &str = "target_id";
+pub const RESULT: &str = "result"; // success / failure
+pub const STATUS: &str = "status"; // success / failure
+
+pub const SUCCESS: &str = "success";
+pub const FAILURE: &str = "failure";
 
 pub static AUDIT_FAILED_MESSAGES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(

crates/obs/src/metrics/bucket.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// bucket level s3 metric descriptor
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, new_histogram_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, new_histogram_md, subsystems};
 use std::sync::LazyLock;
 
 pub static BUCKET_API_TRAFFIC_SENT_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {

crates/obs/src/metrics/bucket_replication.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// Bucket copy metric descriptor
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 /// Bucket level replication metric descriptor

crates/obs/src/metrics/cluster_config.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// Metric descriptors related to cluster configuration
-use crate::metrics::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 
 use std::sync::LazyLock;
 

crates/obs/src/metrics/cluster_erasure_set.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// Erasure code set related metric descriptors
-use crate::metrics::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 /// The label for the pool ID

crates/obs/src/metrics/cluster_health.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// Cluster health-related metric descriptors
-use crate::metrics::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 pub static HEALTH_DRIVES_OFFLINE_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {

crates/obs/src/metrics/cluster_iam.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// IAM related metric descriptors
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, subsystems};
 use std::sync::LazyLock;
 
 pub static LAST_SYNC_DURATION_MILLIS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {

crates/obs/src/metrics/cluster_notification.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// Notify the relevant metric descriptor
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, subsystems};
 use std::sync::LazyLock;
 
 pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {

crates/obs/src/metrics/cluster_usage.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// Descriptors of metrics related to cluster object and bucket usage
-use crate::metrics::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 /// Bucket labels

crates/obs/src/metrics/entry/descriptor.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::metrics::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
+use crate::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
 use std::collections::HashSet;
 
 /// MetricDescriptor - Metric descriptors

crates/obs/src/metrics/entry/mod.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::metrics::{MetricDescriptor, MetricName, MetricNamespace, MetricSubsystem, MetricType};
+use crate::{MetricDescriptor, MetricName, MetricNamespace, MetricSubsystem, MetricType};
 
 pub(crate) mod descriptor;
 pub(crate) mod metric_name;
@@ -76,7 +76,7 @@ pub fn new_histogram_md(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::metrics::subsystems;
+    use crate::subsystems;
 
     #[test]
     fn test_new_histogram_md() {

crates/obs/src/metrics/entry/subsystem.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::metrics::entry::path_utils::format_path_to_metric_name;
+use crate::entry::path_utils::format_path_to_metric_name;
 
 /// The metrics subsystem is a subgroup of metrics within a namespace
 /// The metrics subsystem, which represents a subgroup of metrics within a namespace
@@ -204,8 +204,8 @@ pub mod subsystems {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::metrics::MetricType;
-    use crate::metrics::{MetricDescriptor, MetricName, MetricNamespace};
+    use crate::MetricType;
+    use crate::{MetricDescriptor, MetricName, MetricNamespace};
 
     #[test]
     fn test_metric_subsystem_formatting() {

crates/obs/src/metrics/ilm.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// ILM-related metric descriptors
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 pub static ILM_EXPIRY_PENDING_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {

crates/obs/src/metrics/logger_webhook.rs

@@ -15,7 +15,7 @@
 #![allow(dead_code)]
 
 /// A descriptor for metrics related to webhook logs
-use crate::metrics::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
+use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 
 /// Define label constants for webhook metrics