compressor/zstd: modernize ZSTD API usage in ZstdCompressor

Update ZstdCompressor to use the modern ZSTD compression context API
and improve error handling:

- Replace deprecated ZSTD_createCStream() with ZSTD_createCCtx()
- Use ZSTD_CCtx_reset() with ZSTD_reset_session_and_parameters flag
  instead of separate ZSTD_reset_session_only and ZSTD_CCtx_refCDict()
  calls. This consolidates session and parameter reset into a single
  operation.
- Add proper return value checking for all ZSTD API calls using
  ZSTD_isError() to catch compression failures
- Ensure proper cleanup with ZSTD_freeCCtx() on all error paths to
  prevent memory leaks
- Update corresponding free function from ZSTD_freeCStream() to
  ZSTD_freeCCtx()

These changes align with ZSTD's recommended API usage patterns and
improve robustness by properly handling potential compression errors.

Fixes: https://tracker.ceph.com/issues/73522
Signed-off-by: Edwin Rodriguez <[email protected]>

Author: edwinzrodriguez

src/compressor/zstd/ZstdCompressor.h

@@ -28,8 +28,25 @@ class ZstdCompressor : public Compressor {
   ZstdCompressor(CephContext *cct) : Compressor(COMP_ALG_ZSTD, "zstd"), cct(cct) {}
 
   int compress(const ceph::buffer::list &src, ceph::buffer::list &dst, std::optional<int32_t> &compressor_message) override {
-    ZSTD_CStream *s = ZSTD_createCStream();
-    ZSTD_initCStream_srcSize(s, cct->_conf->compressor_zstd_level, src.length());
+    ZSTD_CCtx *s = ZSTD_createCCtx();
+    if (!s) {
+      return -ENOMEM;
+    }
+    size_t res = ZSTD_CCtx_reset(s, ZSTD_reset_session_and_parameters);
+    if (ZSTD_isError(res)) {
+      ZSTD_freeCCtx(s);
+      return -EINVAL;
+    }
+    res = ZSTD_CCtx_setParameter(s, ZSTD_c_compressionLevel, cct->_conf->compressor_zstd_level);
+    if (ZSTD_isError(res)) {
+      ZSTD_freeCCtx(s);
+      return -EINVAL;
+    }
+    res = ZSTD_CCtx_setPledgedSrcSize(s, src.length());
+    if (ZSTD_isError(res)) {
+      ZSTD_freeCCtx(s);
+      return -EINVAL;
+    }
     auto p = src.begin();
     size_t left = src.length();
 
@@ -49,12 +66,13 @@ class ZstdCompressor : public Compressor {
       ZSTD_EndDirective const zed = (left==0) ? ZSTD_e_end : ZSTD_e_continue;
       size_t r = ZSTD_compressStream2(s, &outbuf, &inbuf, zed);
       if (ZSTD_isError(r)) {
-	return -EINVAL;
+        ZSTD_freeCCtx(s);
+        return -EINVAL;
       }
     }
     ceph_assert(p.get_remaining() == 0);
 
-    ZSTD_freeCStream(s);
+    ZSTD_freeCCtx(s);
 
     // prefix with decompressed length
     ceph::encode((uint32_t)src.length(), dst);