C++20 port 1/14: build, ownership core and allocation audit (#3065)

* Enable C++20 alongside C11 in the build Add CXX to the project with C++20 (no extensions) and build C++ sources with -Wextra -fno-exceptions -fno-rtti -fno-threadsafe-statics on top of the existing warning flags. Give every toolchain a C++ compiler: the zig Windows toolchain, the static macOS/Windows CI jobs, the static musl image (clang++ plus libstdc++ headers) and the RPM spec. * Add C++20 ownership and allocation core Header-only primitives for the no-heap-after-startup design: Result/Status, Region, ObjectPool with PoolPtr (std::unique_ptr with a pool deleter), generation-checked Handle and asynchronous close support, intrusive RefPtr, FixedBuffer, MonotonicArena, FixedHashMap and FixedLru, IntrusiveList, and a bounded FixedHeap for third-party allocator hooks. Unit tests cover each primitive (test_core_*). scripts/check_no_heap.py rejects heap allocation in C++ sources under src/ and runs as a ctest. clang-tidy now also analyses C++ sources and the core headers. * Add heap allocation audit for relayed connections alloc_counter is a preloadable malloc counter (glibc, macOS) that writes a snapshot on SIGUSR2. measure_alloc.py warms ss-local and ss-server up, snapshots both, relays more connections and reports allocations per connection. It runs report-only as the alloc_audit_report ctest. interop.peers() now yields the started processes. * Document the C++20 port and allocation baseline docs/cpp-core.md describes the language profile, the core primitives, the phase plan and the measured C baseline (77-89 allocations per connection in ss-server, 27-39 in ss-local). The compatibility contract now keeps the C11 public header instead of a C11 code base. CLAUDE.md drops stale libev, submodule and gnu99 references. * Fix CI for the C++ core: libstdc++ headers, static unwinder, newer clang - docker/static: Alpine's libstdc++ headers come with g++, and the C++ tests' static libstdc++ needs an unwinder that compiler-rt lacks; add g++ and llvm-libunwind. - FixedBuffer: drop the unused kCapacity (clang-tidy 18 reports it as a dynamically initialized header static). - ss-server on Windows: manager_addr was only read in POSIX-only code, which newer clang rejects (-Wunused-but-set-global); say that the option is ignored there. * CI: allow running the test and build workflows by hand Stacked PRs target their parent branch, which these workflows do not trigger on; workflow_dispatch lets each layer run the full matrix. * CI: let clang-tidy ignore GCC-only warning flags The compile database comes from a GCC build, whose flags (for example -Wno-class-memaccess) clang-tidy does not know and reports as errors.
2026-10-28 10:44:43 +08:00 · 2026-10-15 09:45:54 +08:00
1 parent 553cdc7945
commit bcb3731968
@@ -1,7 +1,10 @@
 # clang-tidy configuration for shadowsocks-c project sources (src/ only).
 # Bundled sources under third_party/ and src/blake3/ are third-party code and
 # are excluded by the file regex used in CI:
-#   run-clang-tidy -p build '/src/[^/]+\.c$'
+#   run-clang-tidy -p build '/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_core_[^/]+\.cpp)$'
+#
+# The header-only C++ core (src/core/*.hpp) is analysed through the
+# test_core_* translation units that include it.
 #
 # The check set is intentionally conservative: the full clang static analyzer
 # plus the bugprone/cert checks that are actionable for this codebase.
@@ -33,4 +36,4 @@ Checks: >
  -clang-analyzer-security.insecureAPI.strcpy,
  -clang-analyzer-security.insecureAPI.DeprecatedOrUnsafeBufferHandling
 WarningsAsErrors: ''
-HeaderFilterRegex: '.*/src/[^/]+\.h$'
+HeaderFilterRegex: '.*/src/([^/]+\.h|core/[^/]+\.hpp)$'
@@ -5,6 +5,7 @@ on:
    branches: [master]
  pull_request:
    branches: [master]
+  workflow_dispatch:

 jobs:
  build:
@@ -94,7 +94,7 @@ jobs:
      - uses: actions/checkout@v6
      - name: Build Clang static-only installation
        run: |
-          cmake -S . -B build-static -DCMAKE_C_COMPILER=clang -DSS_DEPENDENCY_MODE=bundled -DWITH_STATIC=ON -DSS_BUILD_STATIC_LIBRARY=ON -DSS_BUILD_SHARED_LIBRARY=OFF
+          cmake -S . -B build-static -DCMAKE_C_COMPILER=clang -DCMAKE_CXX_COMPILER=clang++ -DSS_DEPENDENCY_MODE=bundled -DWITH_STATIC=ON -DSS_BUILD_STATIC_LIBRARY=ON -DSS_BUILD_SHARED_LIBRARY=OFF
          cmake --build build-static --parallel 4
          ctest --test-dir build-static -L 'unit|vendor' --output-on-failure --no-tests=error
          cmake --install build-static --prefix "$PWD/install-static"
@@ -135,7 +135,7 @@ jobs:
            mingw-w64-clang-x86_64-python
      - name: Build Clang static-only installation
        run: |
-          cmake -S . -B build-static -G Ninja -DCMAKE_C_COMPILER=clang -DSS_DEPENDENCY_MODE=bundled -DWITH_STATIC=ON -DSS_BUILD_STATIC_LIBRARY=ON -DSS_BUILD_SHARED_LIBRARY=OFF
+          cmake -S . -B build-static -G Ninja -DCMAKE_C_COMPILER=clang -DCMAKE_CXX_COMPILER=clang++ -DSS_DEPENDENCY_MODE=bundled -DWITH_STATIC=ON -DSS_BUILD_STATIC_LIBRARY=ON -DSS_BUILD_SHARED_LIBRARY=OFF
          cmake --build build-static --parallel 4
          ctest --test-dir build-static -L 'unit|vendor' --output-on-failure --no-tests=error
          cmake --install build-static --prefix "$PWD/install-static"
@@ -5,6 +5,7 @@ on:
    branches: [master]
  pull_request:
    branches: [master]
+  workflow_dispatch:

 jobs:
  lint:
@@ -124,7 +125,7 @@ jobs:
        run: |
          set -o pipefail
          tidy_status=0
-          run-clang-tidy-18 -quiet -p build "$PWD/src/[^/]+\.c$" 2>&1 | tee tidy.log || tidy_status=$?
+          run-clang-tidy-18 -quiet -p build -extra-arg=-Wno-unknown-warning-option "$PWD/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_core_[^/]+\.cpp)$" 2>&1 | tee tidy.log || tidy_status=$?
          grep 'warning:' tidy.log | sort -u > warnings.txt || true
          count="$(wc -l < warnings.txt)"
          echo "clang-tidy warnings: $count (max allowed: $MAX_WARNINGS)"
@@ -4,78 +4,82 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co

 ## Project Overview

-shadowsocks-libev is a lightweight SOCKS5 proxy written in pure C. Version 3.3.6, licensed under GPLv3.
+shadowsocks-c (formerly shadowsocks-libev) is a lightweight SOCKS5 proxy. Version 3.3.6, licensed under GPLv3.
+The sources are C11 and are being ported to C++20 module by module (see `docs/cpp-core.md`); the public
+`src/shadowsocks.h` stays a C header with a stable ABI.

 ## Build Commands

 ### CMake (sole build system)

 ```bash
-git submodule update --init --recursive
-mkdir -p build && cd build
-cmake ..
-make
-sudo make install
+cmake --preset bundled          # or: cmake -S . -B build -G Ninja
+cmake --build --preset bundled
+ctest --preset bundled
 ```

-On macOS, CMake should auto-detect library paths. If needed, specify paths:
-```bash
-cmake .. -DCMAKE_PREFIX_PATH="/usr/local/opt/mbedtls;/usr/local/opt/libsodium"
-```
+Bundled mode (the default, `SS_DEPENDENCY_MODE=bundled`) builds every dependency from the pinned archives in
+`third_party/dist` with no network access. `SS_DEPENDENCY_MODE=system` uses distro libraries instead.
+There are no git submodules.

-CMake outputs binaries to `build/bin/` (static) and `build/shared/bin/` (shared).
+Binaries land in `<build>/bin/`, libraries in `<build>/lib/`.

 ### Build Dependencies

- cmake (>= 3.2), a C compiler (gcc or clang), pkg-config
- libmbedtls, libsodium (>= 1.0.4), libpcre3, libev, libc-ares
- asciidoc + xmlto (documentation only)
+- CMake >= 3.20, C11 and C++20 compilers (GCC >= 11, Clang >= 14, Apple Clang >= 15), Ninja for the presets
+- System mode only: Mbed TLS 2.x/3.x, libsodium >= 1.0.4, PCRE2 (8-bit), libuv, c-ares
+- Doxygen (documentation only)

 ### CMake Options

- `-DWITH_EMBEDDED_SRC=OFF`: use system libcork/libipset/libbloom instead of bundled submodules
- `-DWITH_DOC_MAN=OFF`: skip man page generation
- `-DENABLE_CONNMARKTOS=ON`: Linux netfilter conntrack QoS support
- `-DENABLE_NFTABLES=ON`: nftables firewall integration
- `-DDISABLE_SSP=ON`: disable stack protector
- `-DBUILD_TESTING=OFF`: disable unit tests
+- `-DSS_MINIMAL=ON`: no regex, plugins, manager or legacy stream ciphers
+- `-DWITH_STATIC=OFF`: link third-party dependencies dynamically
+- `-DSS_BUILD_EXECUTABLES / SS_BUILD_STATIC_LIBRARY / SS_BUILD_SHARED_LIBRARY`: choose outputs
+- `-DWITH_DOC_MAN=ON` / `-DWITH_DOC_HTML=ON`: build documentation
+- `-DENABLE_CONNMARKTOS=ON`, `-DENABLE_NFTABLES=ON`: Linux netfilter features
+- `-DENABLE_SANITIZERS=ON`, `-DENABLE_COVERAGE=ON`, `-DDISABLE_SSP=ON`, `-DBUILD_TESTING=OFF`

 ## Testing

 ### Unit Tests (CTest)

 ```bash
-cd build
-ctest --output-on-failure
+ctest --test-dir build -LE memcheck --output-on-failure
 ```

-10 unit test modules cover: base64, buffer, crypto, json, jconf, cache, ppbloom, rule, netutils, utils.
+C unit tests (`tests/test_*.c`) compile the specific `src/*.c` files they need (see `ss_add_test` in
+`tests/CMakeLists.txt`). C++ core tests are `tests/test_core_*.cpp` using `tests/test_helpers.hpp`.
+Labels: `unit`, `vendor`, `integration`, `memcheck` (Linux + valgrind).

 ### Integration Tests

-Integration tests use Python and require `curl` and `dig` to be available:
 ```bash
-bash tests/test.sh
+python3 tests/interop.py --self --bin build/bin          # TCP/UDP relay, all AEAD/2022 methods
+python3 tests/test.py --bin build/bin/ -c tests/aes-gcm.json   # one cipher via curl/dig (needs network)
+python3 tests/test_cli.py --bin-dir build/bin           # CLI contract: help/version/exit codes
+python3 tests/stress_test.py --bin build/bin/ --size 10
 ```

-The test harness (`tests/test.py`) starts ss-server, ss-local, and ss-tunnel locally, then runs curl through the SOCKS5 proxy and dig through the tunnel. Each test config in `tests/*.json` exercises a different cipher.
+### Allocation audit

-Run a single cipher test:
 ```bash
-python tests/test.py --bin build/bin/ -c tests/aes-gcm.json
+python3 tests/alloc_audit/measure_alloc.py --bin build/bin --shim build/lib/libss_alloc_audit.dylib
 ```

+Reports heap allocations per relayed connection (C baseline numbers are in `docs/cpp-core.md`).
+
 ## Code Formatting

 Uses **uncrustify** with the config at `.uncrustify.cfg`. Key settings: 4-space indent, no tabs, 120-column width, K&R brace style (braces on same line).

 ## Code Quality Tooling

- **clang-tidy**: config in `.clang-tidy` (clang-analyzer + bugprone + cert checks, scoped to `src/`). The build exports `compile_commands.json` automatically. Run locally:
+- **clang-tidy**: config in `.clang-tidy`. The build exports `compile_commands.json` automatically. Run locally:
  ```bash
-  run-clang-tidy -quiet -p build '/src/[^/]+\.c$'
+  run-clang-tidy -quiet -p build '/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_core_[^/]+\.cpp)$'
  ```
  On macOS with Homebrew LLVM, add `-extra-arg="-isysroot$(xcrun --show-sdk-path)"`. CI (`clang-tidy` job in `tests.yml`) enforces a warning-count ratchet via `MAX_WARNINGS` — lower it when fixing findings; never raise it without justification.
+- **No-heap lint**: `scripts/check_no_heap.py --self-test` (also a ctest) rejects heap allocation in C++ sources under `src/`.
 - **Sanitizers**: `cmake .. -DENABLE_SANITIZERS=ON` builds with ASan + UBSan. CI runs ctest and the stress test under sanitizers on every PR.
 - **Coverage**: `cmake .. -DENABLE_COVERAGE=ON`, run tests, then `make coverage` (needs lcov). HTML report lands in `build/coverage-html/`. CI uploads it as the `coverage-html` artifact.

@@ -91,44 +95,41 @@ Each binary is compiled with a module define that controls conditional compilati
 | `ss-server` | `MODULE_REMOTE` | Server-side proxy |
 | `ss-tunnel` | `MODULE_TUNNEL` | Port forwarding tunnel (implies `MODULE_LOCAL`) |
 | `ss-redir` | `MODULE_REDIR` | Transparent proxy via iptables (Linux only, implies `MODULE_LOCAL`) |
-| `ss-manager` | `MODULE_MANAGER` | Multi-server manager daemon |
+| `ss-manager` | `MODULE_MANAGER` | Multi-server manager daemon (Linux only by default) |

-A shared library `libshadowsocks-libev` is also built from the ss-local sources with `-DLIB_ONLY`. Its public API is in `src/shadowsocks.h`.
+A library `libshadowsocks-c` (legacy names `libshadowsocks-libev.*` are kept) is built from the ss-local sources with `-DLIB_ONLY`. Its public API is in `src/shadowsocks.h`.

 ### Source Organization (`src/`)

 **Shared by all binaries:**
- `utils.c` - logging, system utilities
+- `utils.c` - logging, system utilities, `ss_malloc`/`ss_free`
 - `jconf.c` / `json.c` - JSON config file parsing
 - `netutils.c` - network address utilities
- `cache.c` - hash-based LRU connection cache
+- `cache.c` - uthash-based connection cache (UDP relay, AEAD-2022 tables)
 - `udprelay.c` - UDP relay implementation (shared, but uses `#ifdef MODULE_*` for per-binary behavior)
+- `ss_event.c` / `ss_event.h` - event loop API over libuv
+- `resolv.c` - asynchronous DNS via c-ares

-**Crypto layer** (two parallel implementations behind a common `crypto_t` interface):
- `crypto.c` / `crypto.h` - crypto initialization, key derivation (HKDF), buffer management. Defines `crypto_t` with function pointers for encrypt/decrypt.
- `stream.c` - stream cipher implementation (CFB mode via mbedTLS)
- `aead.c` - AEAD cipher implementation (AES-GCM via mbedTLS, ChaCha20-Poly1305 via libsodium)
- `ppbloom.c` - ping-pong bloom filter for nonce replay detection
+**Crypto layer** (behind a common `crypto_t` function-pointer interface):
+- `crypto.c` / `crypto.h` - crypto initialization, key derivation (HKDF), `buffer_t` management
+- `stream.c` - legacy stream ciphers
+- `aead.c` - AEAD ciphers (AES-GCM via Mbed TLS/libsodium, ChaCha20-Poly1305 via libsodium)
+- `aead2022.c` / `aead2022_udp.c` - Shadowsocks 2022 (SIP022), BLAKE3 from `src/blake3`
+- `ppbloom.c` - ping-pong bloom filter for salt replay detection (`third_party/bloom`)

-**ACL (Access Control Lists):**
- `acl.c` / `rule.c` - IP/domain-based routing rules using libipset
+**ACL:** `acl.c` / `rule.c` / `ipset.c` - IP prefix sets and PCRE2/literal domain rules.

-**Plugin support:**
- `plugin.c` - SIP003 plugin subprocess management
+**Plugins:** `plugin.c` / `process.c` - SIP003 plugin subprocess management.

-### Bundled Submodules
-
-Three git submodules in the repo root (can be replaced with system libs via `-DWITH_EMBEDDED_SRC=OFF`):
- `libcork/` - data structures (dllist, hash-table, buffers)
- `libipset/` - IP set operations for ACL
- `libbloom/` - bloom filter implementation
+**C++ core:** `src/core/*.hpp` - header-only ownership and allocation primitives (pools, handles, fixed
+buffers, arena, fixed hash map/LRU, intrusive list, bounded heap). See `docs/cpp-core.md`.

 ### Event Loop

-All binaries use **libev** for async I/O. The connection lifecycle follows stages defined in `src/common.h`: `STAGE_INIT` -> `STAGE_HANDSHAKE` -> `STAGE_RESOLVE` -> `STAGE_STREAM` -> `STAGE_STOP`. Each binary defines its own `listen_ctx_t`, `server_t`, and `remote_t` structs (note: "server" in `local.h` means the local-side connection, "remote" means the ss-server side).
+All binaries use the `ss_event` API (libuv backend: kqueue/epoll/IOCP). The connection lifecycle follows stages defined in `src/common.h`: `STAGE_INIT` -> `STAGE_HANDSHAKE` -> `STAGE_RESOLVE` -> `STAGE_STREAM` -> `STAGE_STOP`. Each binary defines its own `listen_ctx_t`, `server_t`, and `remote_t` structs (note: "server" in `local.h` means the local-side connection, "remote" means the ss-server side).

 ### Compiler Flags

-Default flags from `CMakeLists.txt`: `-g -O2 -Wall -Werror -Wno-deprecated-declarations -fno-strict-aliasing -std=gnu99 -D_GNU_SOURCE`
+From `cmake/ProjectOptions.cmake`: `-Wall -Werror -Wno-deprecated-declarations -fno-strict-aliasing -fstack-protector-strong`, plus `_GNU_SOURCE` on Linux. C is C11 with GNU extensions. C++ is C++20 without extensions and adds `-Wextra -fno-exceptions -fno-rtti -fno-threadsafe-statics`.

-The `-Werror` flag means all warnings are errors - new code must compile warning-free.
+`-Werror` means all warnings are errors — new code must compile warning-free.
@@ -6,7 +6,7 @@ set(PROJECT_VERSION "3.3.6")
 set(PROJECT_DESC "a lightweight secured socks5 proxy")
 set(PROJECT_URL "https://shadowsocks.org")
 set(PROJECT_ISSUES_URL "https://github.com/shadowsocks/shadowsocks-libev")
-project(${PROJECT_NAME} VERSION ${PROJECT_VERSION} LANGUAGES C)
+project(${PROJECT_NAME} VERSION ${PROJECT_VERSION} LANGUAGES C CXX)
 if(MSVC)
    message(FATAL_ERROR "MSVC is not supported yet. Use MinGW-w64 (UCRT64) or the Zig Windows toolchain; see docs/modernization.md.")
 endif()
@@ -23,6 +23,13 @@ set(CMAKE_C_STANDARD 11)
 set(CMAKE_C_STANDARD_REQUIRED ON)
 set(CMAKE_C_EXTENSIONS ON)

+# The C++20 core (src/core) is being introduced module by module; see
+# docs/cpp-core.md. It is built without exceptions or RTTI, so it adds
+# no unwinding tables or typeinfo to the static binaries.
+set(CMAKE_CXX_STANDARD 20)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+set(CMAKE_CXX_EXTENSIONS OFF)
+
 option(ENABLE_SANITIZERS "Build with AddressSanitizer and UndefinedBehaviorSanitizer" OFF)
 option(ENABLE_COVERAGE "Build with coverage instrumentation" OFF)
 option(SS_WARNINGS_AS_ERRORS "Treat project warnings as errors" ON)
@@ -25,6 +25,10 @@ function(ss_project_options target)
        endif()
    else()
        target_compile_options(${target} PRIVATE -Wall -Wno-deprecated-declarations -fno-strict-aliasing)
+        # C++ sources: no exceptions or RTTI, and no locks around function
+        # statics (the event loop is single-threaded).
+        target_compile_options(${target} PRIVATE
+            $<$<COMPILE_LANGUAGE:CXX>:-Wextra -fno-exceptions -fno-rtti -fno-threadsafe-statics>)
        if(SS_WARNINGS_AS_ERRORS)
            target_compile_options(${target} PRIVATE -Werror)
        endif()
@@ -1,9 +1,10 @@
-# Cross-compile the C project from a Unix host without Docker or a MinGW SDK.
+# Cross-compile the project from a Unix host without Docker or a MinGW SDK.
 # cmake -S . -B build-windows -DCMAKE_TOOLCHAIN_FILE=cmake/toolchains/zig-windows.cmake
 set(CMAKE_SYSTEM_NAME Windows)
 set(CMAKE_SYSTEM_PROCESSOR AMD64)
 find_program(SS_ZIG_EXECUTABLE NAMES zig REQUIRED)
 set(CMAKE_C_COMPILER "${SS_ZIG_EXECUTABLE}" cc -target x86_64-windows-gnu)
+set(CMAKE_CXX_COMPILER "${SS_ZIG_EXECUTABLE}" c++ -target x86_64-windows-gnu)
 # CMake's archive tools accept an executable path, not an argv list.
 file(MAKE_DIRECTORY "${CMAKE_BINARY_DIR}/zig-tools")
 foreach(tool ar ranlib)
@@ -2,8 +2,8 @@ ARG ALPINE_IMAGE=alpine:3.24
 FROM ${ALPINE_IMAGE} AS build
 ARG ALPINE_MIRROR=https://dl-cdn.alpinelinux.org/alpine
 RUN sed -i "s|https://dl-cdn.alpinelinux.org/alpine|${ALPINE_MIRROR}|g" /etc/apk/repositories \
- && apk add --no-cache clang lld make musl-dev compiler-rt cmake linux-headers python3 binutils file
-ENV CC=clang
+ && apk add --no-cache clang lld make musl-dev compiler-rt llvm-libunwind-static g++ libstdc++-dev cmake linux-headers python3 binutils file
+ENV CC=clang CXX=clang++
 WORKDIR /src
 COPY . .
 ARG SS_MINIMAL=OFF
@@ -11,7 +11,7 @@ RUN --network=none cmake -S . -B /build \
      -DCMAKE_BUILD_TYPE=Release \
      -DSS_DEPENDENCY_MODE=bundled -DWITH_STATIC=ON \
      -DSS_BUILD_STATIC_LIBRARY=OFF -DSS_BUILD_SHARED_LIBRARY=OFF \
-      -DSS_MINIMAL=${SS_MINIMAL} -DCMAKE_EXE_LINKER_FLAGS="-static -fuse-ld=lld --rtlib=compiler-rt" \
+      -DSS_MINIMAL=${SS_MINIMAL} -DCMAKE_EXE_LINKER_FLAGS="-static -fuse-ld=lld --rtlib=compiler-rt --unwindlib=libunwind" \
      -DCMAKE_INSTALL_PREFIX=/out \
 && cmake --build /build --parallel 4 \
 && ctest --test-dir /build -L 'unit|vendor' --output-on-failure --no-tests=error \
@@ -0,0 +1,97 @@
+# C++20 core and allocation policy
+
+Status: phase 1 of an incremental port. The ownership and allocation
+primitives in `src/core/` exist and are tested; no program code uses them yet.
+All binaries are still built from the C sources.
+
+## Goals
+
+- Move `src/` from C11 to C++20, one module at a time, with every step
+  keeping the CLI, the JSON configuration, protocol behaviour and the public
+  `shadowsocks.h` C ABI unchanged.
+- Make ownership explicit. Each object has exactly one owner, a
+  `std::unique_ptr`. Non-owning references are generation-checked handles
+  that go stale instead of dangling.
+- No heap allocation after startup. Connections, buffers, cipher contexts,
+  DNS queries and UDP sessions come from fixed-capacity pools reserved once
+  at startup. The steady-state data path does not call `malloc`. When a pool
+  is full, new work is refused rather than allocated.
+
+## Language profile
+
+- C++20, `-fno-exceptions -fno-rtti -fno-threadsafe-statics`, warnings as
+  errors. Errors are returned as `ss::Result<T>` (`src/core/result.hpp`), which
+  implements a subset of `std::expected`.
+- Only header-only parts of the standard library are used (`<array>`,
+  `<span>`, `<optional>`, `<variant>`, `<memory>`'s `std::unique_ptr`). This
+  keeps the binaries and the static library free of libstdc++/libc++ symbols,
+  so C consumers of `libshadowsocks-c.a` do not need a C++ runtime.
+- `scripts/check_no_heap.py` (a ctest) rejects `new`, `malloc`, allocating
+  containers, `std::string`, `std::function` and `std::shared_ptr` in C++
+  sources under `src/`. A startup reservation that really must allocate is
+  marked `// heap-ok: <reason>`.
+- blake3 (`src/blake3`) and `third_party/bloom` stay C.
+- Minimum compilers: GCC 11, Clang 14, Apple Clang 15.
+
+## Primitives (`src/core/`)
+
+| Header | Purpose |
+|---|---|
+| `result.hpp` | `Result<T>`, `Status`, `Errc`, `SS_TRY`. |
+| `region.hpp` | `Region`, the only startup `calloc` in the core. Also ASan poison helpers. |
+| `pool.hpp` | `ObjectPool<T>`, `PoolPtr<T>` (`std::unique_ptr` with a pool deleter), `Handle<T>`. Types that embed libuv handles model `AsyncClosable`: their slot is recycled only after the asynchronous close finishes. |
+| `ref.hpp` | `RefCounted<T>`/`RefPtr<T>`, intrusive shared ownership for the few genuinely shared objects (AEAD-2022 UDP sessions). Used instead of `std::shared_ptr` because its control block cannot be pooled portably. |
+| `buffer.hpp` | `FixedBuffer<N>`: inline bytes with `std::span` views; replaces `buffer_t`. |
+| `arena.hpp` | `MonotonicArena` for configuration data that lives as long as the process. |
+| `fixed_map.hpp` | `FixedHashMap` (open addressing, backward-shift deletion) and `FixedLru`; replace uthash and `cache.c`. |
+| `intrusive_list.hpp` | `IntrusiveList`/`ListNode`; replaces `ss_list`. Nodes unlink themselves when destroyed. |
+| `fixed_heap.hpp` | Bounded size-class allocator to hand to libuv, c-ares and Mbed TLS through their allocator hooks. |
+
+## Phases
+
+1. Build and core (this change): C++ enabled in CMake, `src/core/`, unit tests
+   (`test_core_*`), the no-heap lint and the allocation audit.
+2. Mechanical C-to-C++ conversion, leaf modules first.
+3. Event loop wrapper over libuv.
+4. Crypto: cipher contexts held by value, dispatched with `std::variant`.
+5. Services: cache, lists, resolver, ACL and configuration on the new primitives.
+6. One TCP relay engine shared by ss-tunnel, ss-local, ss-redir and ss-server.
+7. UDP relay engine.
+8. Seal the heap after startup; new `max_connections`, `max_udp_sessions`
+   and `max_dns_queries` settings size the pools.
+9. Public API: `shadowsocks.h` kept byte-for-byte, plus a new installed C++
+   header with an RAII handle.
+10. Remove the C shims.
+
+## Allocation audit
+
+`tests/alloc_audit/alloc_counter.c` is a preloadable `malloc` counter
+(glibc and macOS). `tests/alloc_audit/measure_alloc.py` warms ss-local and
+ss-server up, snapshots their counters, relays more connections and reports
+heap allocations per connection. It runs as the `alloc_audit_report` ctest
+(report-only for now). It becomes a hard check, with a budget of zero, once
+the relay is ported.
+
+```sh
+python3 tests/alloc_audit/measure_alloc.py --bin build/bin \
+    --shim build/lib/libss_alloc_audit.dylib --method aes-256-gcm
+```
+
+### C baseline
+
+Commit `553cdc7`, macOS arm64, bundled Release build, 200 connections, each
+echoing 16 KiB through ss-local and ss-server on loopback. The figures
+include allocations made inside libuv, c-ares and Mbed TLS. ss-server
+resolves the destination (`localhost`) for every connection, so its count
+includes a DNS lookup.
+
+| Method | ss-server allocations/connection | ss-local allocations/connection |
+|---|---:|---:|
+| aes-256-gcm | 89.1 | 39.2 |
+| chacha20-ietf-poly1305 | 83.1 | 33.0 |
+| 2022-blake3-aes-256-gcm | 84.0 | 33.0 |
+| aes-256-cfb (stream) | 77.1 | 26.5 |
+| aes-256-gcm, 1-byte payload | 85.0 | 36.0 |
+
+Unstripped Release sizes at the same commit: ss-local 1,086,888 bytes,
+ss-server 1,086,952 bytes, ss-tunnel 542,200 bytes.
@@ -6,12 +6,17 @@ review: https://github.com/shadowsocks/shadowsocks-libev/pull/3050 (merged).

 ## Compatibility contract

-Keep C11 and the existing CLI, JSON configuration, ss:// parsing, SIP003 plugins,
+Keep the existing CLI, JSON configuration, ss:// parsing, SIP003 plugins,
 ACL blacklist/whitelist/outbound precedence, public shadowsocks.h API, legacy
 stream ciphers, AEAD and SIP022 TCP/UDP behavior in the compatibility profile.
 Removing regex or legacy ciphers is explicit opt-in, with configuration errors
 for unavailable functionality. Linux redirection/netfilter remains Linux-only.

+The sources are moving from C11 to C++20 module by module, with an explicit
+ownership model and no heap allocation after startup. The public
+`shadowsocks.h` header stays a C11 header with an unchanged ABI. See
+[C++20 core and allocation policy](cpp-core.md).
+
 ## Work and evidence

 - [x] Baseline: fresh rebuild, unit/integration results, binary/dependency sizes,
@@ -44,7 +49,7 @@ The fresh baseline passes all 13 original unit tests.

 ## Build interface

-Requires CMake 3.20+, a C11 compiler, and a build executor. Presets use Ninja;
+Requires CMake 3.20+, C11 and C++20 compilers, and a build executor. Presets use Ninja;
 manual configuration may use another generator.

 ```
@@ -27,7 +27,7 @@ License:	    GPLv3+
 URL:		    https://github.com/shadowsocks/%{name}
 Source0:	    @SOURCE@

-BuildRequires:	make gcc pcre-devel asciidoc xmlto automake libtool mbedtls-devel libsodium-devel >= 1.0.4 libev-devel c-ares-devel
+BuildRequires:	make gcc gcc-c++ pcre-devel asciidoc xmlto automake libtool mbedtls-devel libsodium-devel >= 1.0.4 libev-devel c-ares-devel
 %if 0%{?suse_version}
 BuildRequires:  libopenssl-devel
 %else
@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Reject heap allocation in the C++ sources under src/.
+
+The C++ core follows a "no heap after startup" policy (docs/modernization.md):
+connections, buffers, crypto contexts and sessions come from fixed pools that
+are reserved once at startup. This lint keeps allocating constructs out of
+C++ code so the policy cannot erode one convenience at a time.
+
+Only C++ files (.cpp, .hpp) are checked; the legacy C sources are ported
+module by module and keep their ss_malloc calls until then.
+
+A line that must allocate (a startup reservation) carries a justification:
+
+    void *base = std::calloc(1, size); // heap-ok: startup reservation
+
+Usage: check_no_heap.py [--root DIR] [--self-test]
+"""
+
+import argparse
+import pathlib
+import re
+import sys
+
+ALLOW_MARKER = "heap-ok:"
+
+# (pattern, explanation). Patterns run on code with comments and string
+# literals blanked out.
+FORBIDDEN = [
+    (re.compile(r"\b(?:std::)?(?:malloc|calloc|realloc|aligned_alloc|posix_memalign|strdup|strndup)\s*\("),
+     "C allocator call"),
+    (re.compile(r"\bss_(?:malloc|realloc|aligned_malloc|strndup)\s*\("), "ss_* allocator call"),
+    (re.compile(r"\b(?:std::)?free\s*\("), "free(); memory comes from pools, not malloc"),
+    # `new T`, `new T[n]`, `::new T` -- but not placement new `new (p) T`.
+    (re.compile(r"(?<![\w.>])(?:::)?new\s+(?!\()[A-Za-z_:]"), "operator new"),
+    (re.compile(r"\bdelete(?:\s*\[\s*\])?\s+[A-Za-z_*(]"), "operator delete"),
+    (re.compile(r"\bstd::(?:make_unique|make_shared|allocate_shared|make_unique_for_overwrite)\b"),
+     "heap-allocating smart pointer factory; acquire from an ObjectPool"),
+    (re.compile(r"\bstd::(?:shared_ptr|weak_ptr)\b"), "std::shared_ptr; use ss::RefPtr over a pool"),
+    (re.compile(r"\bstd::(?:vector|deque|list|forward_list|map|multimap|set|multiset|"
+                r"unordered_map|unordered_multimap|unordered_set|unordered_multiset)\s*<"),
+     "allocating container; use std::array, std::span or src/core containers"),
+    (re.compile(r"\bstd::(?:string|wstring|u8string|stringstream|ostringstream|istringstream|function)\b"),
+     "allocating type; use std::string_view, a FixedBuffer or a function pointer"),
+]
+
+
+def strip_comments_and_strings(text):
+    """Blank out comments and string/char literals, preserving line breaks."""
+    out = []
+    i, n = 0, len(text)
+    while i < n:
+        c = text[i]
+        if text.startswith("//", i):
+            j = text.find("\n", i)
+            j = n if j == -1 else j
+            out.append(" " * (j - i))
+            i = j
+        elif text.startswith("/*", i):
+            j = text.find("*/", i + 2)
+            j = n if j == -1 else j + 2
+            out.append("".join("\n" if ch == "\n" else " " for ch in text[i:j]))
+            i = j
+        elif c in "\"'":
+            j = i + 1
+            while j < n and text[j] != c and text[j] != "\n":
+                j += 2 if text[j] == "\\" else 1
+            j = min(j + 1, n)
+            out.append(c + " " * (j - i - 2) + c if j - i >= 2 else c)
+            i = j
+        else:
+            out.append(c)
+            i += 1
+    return "".join(out)
+
+
+def check_text(text):
+    """Return a list of (line_number, explanation, source_line) violations."""
+    problems = []
+    raw_lines = text.splitlines()
+    code_lines = strip_comments_and_strings(text).splitlines()
+    for number, (raw, code) in enumerate(zip(raw_lines, code_lines), start=1):
+        if ALLOW_MARKER in raw:
+            continue
+        for pattern, why in FORBIDDEN:
+            if pattern.search(code):
+                problems.append((number, why, raw.strip()))
+                break
+    return problems
+
+
+def check_tree(root):
+    src = root / "src"
+    files = sorted(p for p in src.rglob("*") if p.suffix in (".cpp", ".hpp") and "blake3" not in p.parts)
+    failures = 0
+    for path in files:
+        for number, why, line in check_text(path.read_text(encoding="utf-8")):
+            print(f"{path.relative_to(root)}:{number}: {why}: {line}")
+            failures += 1
+    print(f"check_no_heap: {len(files)} C++ files, {failures} violation(s)")
+    return failures == 0
+
+
+def self_test():
+    bad = {
+        "int *p = (int *)malloc(4);": True,
+        "auto *c = new Conn();": True,
+        "auto *c = ::new Conn;": True,
+        "delete c;": True,
+        "std::vector<int> v;": True,
+        "std::string name;": True,
+        "std::function<void()> cb;": True,
+        "auto p = std::make_unique<int>(1);": True,
+        "std::shared_ptr<T> s;": True,
+        "ss_free(ptr);": False,  # not a call pattern we flag (macro wraps free)
+        "free(buf);": True,
+    }
+    good = [
+        "std::construct_at(p, 1);",
+        "::new (static_cast<void *>(p)) T(1);",
+        "Foo(const Foo &) = delete;",
+        "std::string_view name;",
+        "std::array<std::byte, 16> key;",
+        "// std::vector<int> mentioned in a comment",
+        'log("new connection from %s", addr);',
+        "void *b = std::calloc(1, n); // heap-ok: startup reservation",
+        "bool renew_lease = true;",
+        "conn.free_slots();",
+    ]
+    ok = True
+    for line, expect in bad.items():
+        got = bool(check_text(line))
+        if got != expect:
+            print(f"self-test: expected {'violation' if expect else 'clean'} for: {line}")
+            ok = False
+    for line in good:
+        if check_text(line):
+            print(f"self-test: false positive for: {line}")
+            ok = False
+    print("check_no_heap self-test:", "ok" if ok else "FAILED")
+    return ok
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--root", type=pathlib.Path, default=pathlib.Path(__file__).resolve().parent.parent)
+    parser.add_argument("--self-test", action="store_true", help="run the pattern self-test first")
+    args = parser.parse_args()
+    if args.self_test and not self_test():
+        return 1
+    return 0 if check_tree(args.root.resolve()) else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,127 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_ARENA_HPP
+#define SS_CORE_ARENA_HPP
+
+// MonotonicArena: bump allocation for data that lives as long as the
+// process configuration (parsed JSON, CLI strings, ACL entries).
+//
+// The arena reserves one block at startup. Allocations never free
+// individually; reset() discards everything at once (ss-manager uses this
+// per control command). Only trivially destructible types may be placed in
+// an arena, because nothing runs destructors.
+
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <span>
+#include <string_view>
+#include <type_traits>
+#include <utility>
+
+#include "core/region.hpp"
+#include "core/result.hpp"
+
+namespace ss {
+
+class MonotonicArena {
+public:
+    MonotonicArena() noexcept = default;
+    MonotonicArena(const MonotonicArena &)            = delete;
+    MonotonicArena &operator=(const MonotonicArena &) = delete;
+    MonotonicArena(MonotonicArena &&)                 = default;
+    MonotonicArena &operator=(MonotonicArena &&)      = default;
+
+    Status reserve(std::size_t bytes) noexcept
+    {
+        if (!block_.empty()) {
+            return fail(Errc::invalid);
+        }
+        auto region = Region::reserve(bytes);
+        if (!region) {
+            return fail(region.error());
+        }
+        block_ = std::move(*region);
+        used_  = 0;
+        return {};
+    }
+
+    Result<std::span<std::byte> > allocate(std::size_t size, std::size_t align = alignof(std::max_align_t)) noexcept
+    {
+        if (align == 0 || (align & (align - 1)) != 0) {
+            return fail(Errc::invalid);
+        }
+        auto base    = reinterpret_cast<std::uintptr_t>(block_.data());
+        auto current = base + used_;
+        auto aligned = (current + align - 1) & ~(static_cast<std::uintptr_t>(align) - 1);
+        std::size_t offset = static_cast<std::size_t>(aligned - base);
+        if (block_.empty() || offset > block_.size() || size > block_.size() - offset) {
+            return fail(Errc::exhausted);
+        }
+        used_ = offset + size;
+        if (used_ > peak_) {
+            peak_ = used_;
+        }
+        return std::span<std::byte>(block_.data() + offset, size);
+    }
+
+    template <class T, class... Args>
+    Result<T *> make(Args &&...args) noexcept
+    {
+        static_assert(std::is_trivially_destructible_v<T>, "arena objects are never destroyed");
+        auto bytes = allocate(sizeof(T), alignof(T));
+        if (!bytes) {
+            return fail(bytes.error());
+        }
+        return std::construct_at(reinterpret_cast<T *>(bytes->data()), std::forward<Args>(args)...);
+    }
+
+    template <class T>
+    Result<std::span<T> > make_array(std::size_t count) noexcept
+    {
+        static_assert(std::is_trivially_destructible_v<T> && std::is_default_constructible_v<T>);
+        if (count > SIZE_MAX / sizeof(T)) {
+            return fail(Errc::exhausted);
+        }
+        auto bytes = allocate(sizeof(T) * count, alignof(T));
+        if (!bytes) {
+            return fail(bytes.error());
+        }
+        T *first = reinterpret_cast<T *>(bytes->data());
+        for (std::size_t i = 0; i < count; ++i) {
+            std::construct_at(first + i);
+        }
+        return std::span<T>(first, count);
+    }
+
+    // Copy a string into the arena, NUL-terminated so it can be handed to C
+    // APIs. The returned view excludes the terminator.
+    Result<std::string_view> intern(std::string_view text) noexcept
+    {
+        auto bytes = allocate(text.size() + 1, 1);
+        if (!bytes) {
+            return fail(bytes.error());
+        }
+        char *out = reinterpret_cast<char *>(bytes->data());
+        if (!text.empty()) {
+            std::memcpy(out, text.data(), text.size());
+        }
+        out[text.size()] = '\0';
+        return std::string_view(out, text.size());
+    }
+
+    void reset() noexcept { used_ = 0; }
+
+    std::size_t capacity() const noexcept { return block_.size(); }
+    std::size_t used() const noexcept { return used_; }
+    std::size_t peak() const noexcept { return peak_; }
+
+private:
+    Region block_;
+    std::size_t used_ = 0;
+    std::size_t peak_ = 0;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,154 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_BUFFER_HPP
+#define SS_CORE_BUFFER_HPP
+
+// FixedBuffer<N>: a byte buffer with inline, compile-time capacity.
+//
+// Replaces buffer_t (crypto.h) and balloc/brealloc/bprepend. The bytes live
+// inside the object, so a buffer embedded in a pooled connection costs no
+// allocation at all. Instead of growing, writers see how much room is left
+// through writable() and stop reading from the socket when it is full.
+//
+// Layout:  [ consumed | readable (idx .. idx+len) | writable ]
+
+#include <algorithm>
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <span>
+
+namespace ss {
+
+using Bytes      = std::span<std::byte>;
+using ConstBytes = std::span<const std::byte>;
+
+inline ConstBytes as_bytes(const void *data, std::size_t size) noexcept
+{
+    return { static_cast<const std::byte *>(data), size };
+}
+
+template <std::size_t N>
+class FixedBuffer {
+    static_assert(N > 0 && N <= UINT32_MAX);
+
+public:
+    constexpr std::size_t capacity() const noexcept { return N; }
+    constexpr std::size_t size() const noexcept { return len_; }
+    constexpr bool empty() const noexcept { return len_ == 0; }
+    constexpr bool full() const noexcept { return idx_ + len_ == N; }
+
+    ConstBytes readable() const noexcept { return { data_.data() + idx_, len_ }; }
+    Bytes readable() noexcept { return { data_.data() + idx_, len_ }; }
+    Bytes writable() noexcept { return { data_.data() + idx_ + len_, N - idx_ - len_ }; }
+
+    // Mark `n` bytes of writable() as filled.
+    bool commit(std::size_t n) noexcept
+    {
+        if (n > N - idx_ - len_) {
+            return false;
+        }
+        len_ += static_cast<std::uint32_t>(n);
+        return true;
+    }
+
+    // Drop `n` bytes from the front of readable().
+    bool consume(std::size_t n) noexcept
+    {
+        if (n > len_) {
+            return false;
+        }
+        idx_ += static_cast<std::uint32_t>(n);
+        len_ -= static_cast<std::uint32_t>(n);
+        if (len_ == 0) {
+            idx_ = 0;
+        }
+        return true;
+    }
+
+    // Move the readable bytes to the front so writable() is as large as it
+    // can be.
+    void compact() noexcept
+    {
+        if (idx_ != 0) {
+            if (len_ != 0) {
+                std::memmove(data_.data(), data_.data() + idx_, len_);
+            }
+            idx_ = 0;
+        }
+    }
+
+    bool append(ConstBytes bytes) noexcept
+    {
+        if (bytes.size() > N - idx_ - len_) {
+            compact();
+            if (bytes.size() > N - len_) {
+                return false;
+            }
+        }
+        if (!bytes.empty()) {
+            std::memcpy(data_.data() + idx_ + len_, bytes.data(), bytes.size());
+        }
+        len_ += static_cast<std::uint32_t>(bytes.size());
+        return true;
+    }
+
+    // Insert bytes in front of readable(), e.g. a salt or an address header.
+    bool prepend(ConstBytes bytes) noexcept
+    {
+        std::size_t n = bytes.size();
+        if (n > N - len_) {
+            return false;
+        }
+        if (n > idx_) {
+            // Not enough headroom: shift the payload right.
+            std::size_t new_idx = n;
+            if (len_ != 0) {
+                std::memmove(data_.data() + new_idx, data_.data() + idx_, len_);
+            }
+            idx_ = static_cast<std::uint32_t>(new_idx);
+        }
+        idx_ -= static_cast<std::uint32_t>(n);
+        if (n != 0) {
+            std::memcpy(data_.data() + idx_, bytes.data(), n);
+        }
+        len_ += static_cast<std::uint32_t>(n);
+        return true;
+    }
+
+    void clear() noexcept
+    {
+        idx_ = 0;
+        len_ = 0;
+    }
+
+    // Overwrite the contents. Fails, leaving the buffer unchanged, if the
+    // bytes do not fit.
+    bool assign(ConstBytes bytes) noexcept
+    {
+        if (bytes.size() > N) {
+            return false;
+        }
+        clear();
+        return append(bytes);
+    }
+
+    // Wipe the storage, for buffers that held key material or plaintext.
+    void wipe() noexcept
+    {
+        volatile std::byte *p = data_.data();
+        for (std::size_t i = 0; i < N; ++i) {
+            p[i] = std::byte{ 0 };
+        }
+        clear();
+    }
+
+private:
+    std::uint32_t idx_ = 0;
+    std::uint32_t len_ = 0;
+    std::array<std::byte, N> data_;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,229 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_FIXED_HEAP_HPP
+#define SS_CORE_FIXED_HEAP_HPP
+
+// FixedHeap: a bounded malloc replacement for third-party libraries.
+//
+// libuv (uv_replace_allocator), c-ares (ares_library_init_mem) and Mbed TLS
+// (MBEDTLS_PLATFORM_MEMORY) can be pointed at our own allocation functions.
+// Each library gets its own FixedHeap, carved out of one region reserved at
+// startup, so their steady-state allocations stay inside a budget and can
+// never starve each other or the relay pools.
+//
+// Blocks come in power-of-two size classes from 32 bytes to 128 KiB, each
+// with a 16-byte header. Freed blocks go to a per-class free list and are
+// reused; untouched memory is handed out by bumping a pointer. Calls are
+// serialised with a spin lock, because libuv's thread pool may allocate
+// from worker threads.
+
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <utility>
+
+#include "core/region.hpp"
+#include "core/result.hpp"
+
+namespace ss {
+
+class FixedHeap {
+public:
+    struct Stats {
+        std::size_t in_use      = 0; // bytes in live blocks, headers included
+        std::size_t peak        = 0;
+        std::size_t reserved    = 0;
+        std::size_t allocations = 0;
+        std::size_t failures    = 0;
+    };
+
+    FixedHeap() noexcept = default;
+    FixedHeap(const FixedHeap &)            = delete;
+    FixedHeap &operator=(const FixedHeap &) = delete;
+
+    Status reserve(std::size_t budget) noexcept
+    {
+        if (!region_.empty()) {
+            return fail(Errc::invalid);
+        }
+        auto region = Region::reserve(budget, kHeader);
+        if (!region) {
+            return fail(region.error());
+        }
+        region_         = std::move(*region);
+        bump_           = 0;
+        stats_.reserved = region_.size();
+        return {};
+    }
+
+    void *allocate(std::size_t size) noexcept
+    {
+        Guard guard(lock_);
+        return allocate_locked(size);
+    }
+
+    void *allocate_zeroed(std::size_t count, std::size_t size) noexcept
+    {
+        if (size != 0 && count > SIZE_MAX / size) {
+            Guard guard(lock_);
+            ++stats_.failures;
+            return nullptr;
+        }
+        std::size_t total = count * size;
+        Guard guard(lock_);
+        void *p = allocate_locked(total);
+        if (p != nullptr) {
+            std::memset(p, 0, total);
+        }
+        return p;
+    }
+
+    void *reallocate(void *old, std::size_t size) noexcept
+    {
+        if (old == nullptr) {
+            return allocate(size);
+        }
+        if (size == 0) {
+            deallocate(old);
+            return nullptr;
+        }
+        Guard guard(lock_);
+        Header *h          = header_of(old);
+        std::size_t usable = class_size(h->cls) - kHeader;
+        if (size <= usable) {
+            return old;
+        }
+        void *fresh = allocate_locked(size);
+        if (fresh != nullptr) {
+            std::memcpy(fresh, old, usable);
+            deallocate_locked(old);
+        }
+        return fresh;
+    }
+
+    void deallocate(void *p) noexcept
+    {
+        if (p == nullptr) {
+            return;
+        }
+        Guard guard(lock_);
+        deallocate_locked(p);
+    }
+
+    bool owns(const void *p) const noexcept
+    {
+        auto *b = static_cast<const std::byte *>(p);
+        return b >= region_.data() && b < region_.data() + region_.size();
+    }
+
+    Stats stats() const noexcept
+    {
+        Guard guard(lock_);
+        return stats_;
+    }
+
+    static constexpr std::size_t max_request() noexcept { return class_size(kClasses - 1) - kHeader; }
+
+private:
+    static constexpr std::size_t kHeader   = 16;
+    static constexpr unsigned kMinShift    = 5;  // 32-byte blocks
+    static constexpr unsigned kClasses     = 13; // ... up to 128 KiB
+    static constexpr std::uint32_t kMagic  = 0x55aa1357U;
+    static constexpr std::uint32_t kFreed  = 0xdeadf7eeU;
+
+    struct Header {
+        std::uint32_t magic;
+        std::uint32_t cls;
+        void *next_free; // valid only while the block is on a free list
+    };
+    static_assert(sizeof(Header) <= kHeader);
+
+    class Guard {
+    public:
+        explicit Guard(std::atomic_flag &flag) noexcept : flag_(flag)
+        {
+            while (flag_.test_and_set(std::memory_order_acquire)) {
+            }
+        }
+        ~Guard() { flag_.clear(std::memory_order_release); }
+        Guard(const Guard &)            = delete;
+        Guard &operator=(const Guard &) = delete;
+
+    private:
+        std::atomic_flag &flag_;
+    };
+
+    static constexpr std::size_t class_size(unsigned cls) noexcept { return std::size_t{ 1 } << (cls + kMinShift); }
+
+    static Header *header_of(void *p) noexcept
+    {
+        auto *h = reinterpret_cast<Header *>(static_cast<std::byte *>(p) - kHeader);
+        if (h->magic != kMagic || h->cls >= kClasses) {
+            std::abort(); // not ours, or corrupted
+        }
+        return h;
+    }
+
+    void *allocate_locked(std::size_t size) noexcept
+    {
+        if (size == 0) {
+            size = 1;
+        }
+        unsigned cls = 0;
+        while (cls < kClasses && class_size(cls) - kHeader < size) {
+            ++cls;
+        }
+        if (cls == kClasses) {
+            ++stats_.failures;
+            return nullptr;
+        }
+        std::byte *block = nullptr;
+        if (free_[cls] != nullptr) {
+            block = static_cast<std::byte *>(free_[cls]);
+            SS_UNPOISON(block, class_size(cls));
+            free_[cls] = reinterpret_cast<Header *>(block)->next_free;
+        } else if (class_size(cls) <= region_.size() - bump_) {
+            block = region_.data() + bump_;
+            bump_ += class_size(cls);
+        } else {
+            ++stats_.failures;
+            return nullptr;
+        }
+        auto *h      = reinterpret_cast<Header *>(block);
+        h->magic     = kMagic;
+        h->cls       = cls;
+        h->next_free = nullptr;
+        stats_.in_use += class_size(cls);
+        ++stats_.allocations;
+        if (stats_.in_use > stats_.peak) {
+            stats_.peak = stats_.in_use;
+        }
+        return block + kHeader;
+    }
+
+    void deallocate_locked(void *p) noexcept
+    {
+        if (!owns(p)) {
+            std::abort();
+        }
+        Header *h = header_of(p);
+        unsigned cls = h->cls;
+        stats_.in_use -= class_size(cls);
+        h->magic     = kFreed; // a second free of this block aborts in header_of
+        h->next_free = free_[cls];
+        free_[cls]   = h;
+        // Keep the header readable for the free list; poison the payload.
+        SS_POISON(static_cast<std::byte *>(p), class_size(cls) - kHeader);
+    }
+
+    Region region_;
+    std::size_t bump_ = 0;
+    void *free_[kClasses] = {};
+    Stats stats_;
+    mutable std::atomic_flag lock_ = ATOMIC_FLAG_INIT;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,495 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_FIXED_MAP_HPP
+#define SS_CORE_FIXED_MAP_HPP
+
+// Fixed-capacity associative containers. Replace uthash (cache.c,
+// ss_event.c, manager.c) without any per-entry allocation.
+//
+// FixedHashMap<K, V>
+//     Open addressing with linear probing and backward-shift deletion (no
+//     tombstones, so lookups stay short however long the process runs).
+//     The table is sized at reserve() time to at least twice the capacity.
+//     Entries move when others are erased, so never keep a pointer to a
+//     value across an erase.
+//
+// FixedLru<K, V>
+//     Map plus recency order. Values live in a slab and never move, so the
+//     pointer returned by insert() or find() stays valid until that entry
+//     is erased or evicted. Replaces the pseudo-LRU in cache.c.
+//
+// Keys from the network (UDP peer addresses, session IDs) are hashed with a
+// per-table seed. Seed it from a CSPRNG in production code so a peer cannot
+// precompute colliding keys.
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <optional>
+#include <type_traits>
+#include <utility>
+
+#include "core/buffer.hpp"
+#include "core/region.hpp"
+#include "core/result.hpp"
+
+namespace ss {
+
+constexpr std::uint64_t mix64(std::uint64_t x) noexcept
+{
+    x ^= x >> 30;
+    x *= 0xbf58476d1ce4e5b9ULL;
+    x ^= x >> 27;
+    x *= 0x94d049bb133111ebULL;
+    x ^= x >> 31;
+    return x;
+}
+
+inline std::uint64_t hash_bytes(const void *data, std::size_t size, std::uint64_t seed) noexcept
+{
+    const auto *p   = static_cast<const unsigned char *>(data);
+    std::uint64_t h = mix64(seed ^ (size * 0x9e3779b97f4a7c15ULL));
+    while (size >= 8) {
+        std::uint64_t chunk;
+        std::memcpy(&chunk, p, 8);
+        h = mix64(h ^ chunk) * 0x9e3779b97f4a7c15ULL;
+        p += 8;
+        size -= 8;
+    }
+    std::uint64_t tail = 0;
+    if (size != 0) {
+        std::memcpy(&tail, p, size);
+    }
+    return mix64(h ^ tail);
+}
+
+// Inline byte-string key of at most N bytes (e.g. a sockaddr or an
+// AEAD-2022 session ID).
+template <std::size_t N>
+class FixedKey {
+    static_assert(N > 0 && N <= 255);
+
+public:
+    constexpr FixedKey() noexcept = default;
+
+    static std::optional<FixedKey> from(ConstBytes bytes) noexcept
+    {
+        if (bytes.size() > N) {
+            return std::nullopt;
+        }
+        FixedKey key;
+        key.len_ = static_cast<std::uint8_t>(bytes.size());
+        if (!bytes.empty()) {
+            std::memcpy(key.data_.data(), bytes.data(), bytes.size());
+        }
+        return key;
+    }
+
+    ConstBytes bytes() const noexcept { return { data_.data(), len_ }; }
+    std::size_t size() const noexcept { return len_; }
+
+    friend bool operator==(const FixedKey &a, const FixedKey &b) noexcept
+    {
+        return a.len_ == b.len_ && std::memcmp(a.data_.data(), b.data_.data(), a.len_) == 0;
+    }
+
+private:
+    std::array<std::byte, N> data_{};
+    std::uint8_t len_ = 0;
+};
+
+template <class K>
+struct Hasher;
+
+template <class K>
+requires std::is_integral_v<K> || std::is_enum_v<K>
+struct Hasher<K> {
+    std::uint64_t operator()(K key, std::uint64_t seed) const noexcept
+    {
+        return mix64(static_cast<std::uint64_t>(key) ^ mix64(seed));
+    }
+};
+
+template <std::size_t N>
+struct Hasher<FixedKey<N> > {
+    std::uint64_t operator()(const FixedKey<N> &key, std::uint64_t seed) const noexcept
+    {
+        return hash_bytes(key.bytes().data(), key.size(), seed);
+    }
+};
+
+template <class K, class V, class Hash = Hasher<K> >
+class FixedHashMap {
+    static_assert(std::is_nothrow_move_constructible_v<K> && std::is_nothrow_move_constructible_v<V>);
+
+public:
+    FixedHashMap() noexcept = default;
+    FixedHashMap(const FixedHashMap &)            = delete;
+    FixedHashMap &operator=(const FixedHashMap &) = delete;
+    ~FixedHashMap() { clear(); }
+
+    Status reserve(std::uint32_t capacity, std::uint64_t seed = 0) noexcept
+    {
+        if (capacity_ != 0 || capacity == 0 || capacity > (UINT32_MAX >> 2)) {
+            return fail(Errc::invalid);
+        }
+        std::uint32_t slots = 2;
+        while (slots < capacity * 2U) {
+            slots <<= 1;
+        }
+        auto entries = Region::reserve(static_cast<std::size_t>(slots) * sizeof(Entry), alignof(Entry));
+        if (!entries) {
+            return fail(entries.error());
+        }
+        auto used = Region::reserve(slots, 1);
+        if (!used) {
+            return fail(used.error());
+        }
+        entries_  = std::move(*entries);
+        used_     = std::move(*used);
+        mask_     = slots - 1;
+        capacity_ = capacity;
+        seed_     = seed;
+        return {};
+    }
+
+    std::uint32_t size() const noexcept { return size_; }
+    std::uint32_t capacity() const noexcept { return capacity_; }
+    bool empty() const noexcept { return size_ == 0; }
+    bool full() const noexcept { return size_ == capacity_; }
+
+    V *find(const K &key) noexcept
+    {
+        std::uint32_t slot;
+        return locate(key, slot) ? &entry(slot)->value : nullptr;
+    }
+    const V *find(const K &key) const noexcept { return const_cast<FixedHashMap *>(this)->find(key); }
+    bool contains(const K &key) const noexcept { return find(key) != nullptr; }
+
+    // Insert a new key. Fails with Errc::exhausted when full and with
+    // Errc::invalid when the key is already present.
+    template <class... Args>
+    Result<V *> emplace(const K &key, Args &&...args) noexcept
+    {
+        if (capacity_ == 0) {
+            return fail(Errc::exhausted);
+        }
+        std::uint32_t slot;
+        if (locate(key, slot)) {
+            return fail(Errc::invalid);
+        }
+        if (full()) {
+            return fail(Errc::exhausted);
+        }
+        // locate() leaves `slot` at the first empty slot of the probe run.
+        Entry *e = std::construct_at(entry(slot), key, std::forward<Args>(args)...);
+        used_.data()[slot] = std::byte{ 1 };
+        ++size_;
+        return &e->value;
+    }
+
+    std::optional<V> take(const K &key) noexcept
+    {
+        std::uint32_t slot;
+        if (!locate(key, slot)) {
+            return std::nullopt;
+        }
+        std::optional<V> out(std::move(entry(slot)->value));
+        remove_at(slot);
+        return out;
+    }
+
+    bool erase(const K &key) noexcept
+    {
+        std::uint32_t slot;
+        if (!locate(key, slot)) {
+            return false;
+        }
+        remove_at(slot);
+        return true;
+    }
+
+    void clear() noexcept
+    {
+        for (std::uint32_t i = 0; capacity_ != 0 && i <= mask_; ++i) {
+            if (occupied(i)) {
+                std::destroy_at(entry(i));
+                used_.data()[i] = std::byte{ 0 };
+            }
+        }
+        size_ = 0;
+    }
+
+    // Visit every entry in unspecified order. The callback must not insert
+    // or erase.
+    template <class F>
+    void for_each(F &&f)
+    {
+        for (std::uint32_t i = 0; capacity_ != 0 && i <= mask_; ++i) {
+            if (occupied(i)) {
+                f(static_cast<const K &>(entry(i)->key), entry(i)->value);
+            }
+        }
+    }
+
+private:
+    struct Entry {
+        template <class... Args>
+        explicit Entry(const K &k, Args &&...args) : key(k), value(std::forward<Args>(args)...)
+        {
+        }
+        K key;
+        V value;
+    };
+
+    Entry *entry(std::uint32_t slot) const noexcept { return reinterpret_cast<Entry *>(entries_.data()) + slot; }
+    bool occupied(std::uint32_t slot) const noexcept { return used_.data()[slot] != std::byte{ 0 }; }
+    std::uint32_t home(const K &key) const noexcept
+    {
+        return static_cast<std::uint32_t>(Hash{}(key, seed_)) & mask_;
+    }
+
+    // True if found (slot = its position); otherwise slot = first empty slot.
+    bool locate(const K &key, std::uint32_t &slot) const noexcept
+    {
+        if (capacity_ == 0) {
+            return false;
+        }
+        std::uint32_t i = home(key);
+        while (occupied(i)) {
+            if (entry(i)->key == key) {
+                slot = i;
+                return true;
+            }
+            i = (i + 1) & mask_;
+        }
+        slot = i;
+        return false;
+    }
+
+    void remove_at(std::uint32_t hole) noexcept
+    {
+        std::destroy_at(entry(hole));
+        used_.data()[hole] = std::byte{ 0 };
+        --size_;
+        // Backward-shift: pull later members of the probe run into the hole
+        // when the hole lies between their home slot and where they sit.
+        std::uint32_t j = hole;
+        for (;;) {
+            j = (j + 1) & mask_;
+            if (!occupied(j)) {
+                return;
+            }
+            std::uint32_t k = home(entry(j)->key);
+            bool movable    = (hole <= j) ? (k <= hole || k > j) : (k <= hole && k > j);
+            if (!movable) {
+                continue;
+            }
+            std::construct_at(entry(hole), std::move(*entry(j)));
+            used_.data()[hole] = std::byte{ 1 };
+            std::destroy_at(entry(j));
+            used_.data()[j] = std::byte{ 0 };
+            hole = j;
+        }
+    }
+
+    Region entries_;
+    Region used_;
+    std::uint32_t mask_     = 0;
+    std::uint32_t capacity_ = 0;
+    std::uint32_t size_     = 0;
+    std::uint64_t seed_     = 0;
+};
+
+template <class K, class V, class Hash = Hasher<K> >
+class FixedLru {
+public:
+    FixedLru() noexcept = default;
+    FixedLru(const FixedLru &)            = delete;
+    FixedLru &operator=(const FixedLru &) = delete;
+    ~FixedLru() { clear(); }
+
+    Status reserve(std::uint32_t capacity, std::uint64_t seed = 0) noexcept
+    {
+        SS_TRY(index_.reserve(capacity, seed));
+        auto nodes = Region::reserve(static_cast<std::size_t>(capacity) * sizeof(Node), alignof(Node));
+        if (!nodes) {
+            return fail(nodes.error());
+        }
+        nodes_    = std::move(*nodes);
+        capacity_ = capacity;
+        for (std::uint32_t i = 0; i < capacity; ++i) {
+            std::construct_at(node(i))->next = (i + 1 < capacity) ? i + 1 : kNone;
+        }
+        free_ = 0;
+        return {};
+    }
+
+    std::uint32_t size() const noexcept { return index_.size(); }
+    std::uint32_t capacity() const noexcept { return capacity_; }
+    bool empty() const noexcept { return size() == 0; }
+    bool full() const noexcept { return free_ == kNone; }
+
+    // Look up and mark as most recently used.
+    V *find(const K &key) noexcept
+    {
+        std::uint32_t *slot = index_.find(key);
+        if (slot == nullptr) {
+            return nullptr;
+        }
+        touch(*slot);
+        return value(*slot);
+    }
+
+    // Look up without changing the recency order.
+    V *peek(const K &key) noexcept
+    {
+        std::uint32_t *slot = index_.find(key);
+        return slot == nullptr ? nullptr : value(*slot);
+    }
+
+    // Insert as most recently used. Fails with Errc::exhausted when full
+    // (evict with pop_oldest() first) and Errc::invalid for a duplicate key.
+    template <class... Args>
+    Result<V *> emplace(const K &key, Args &&...args) noexcept
+    {
+        if (free_ == kNone) {
+            return fail(Errc::exhausted);
+        }
+        std::uint32_t slot = free_;
+        auto indexed       = index_.emplace(key, slot);
+        if (!indexed) {
+            return fail(indexed.error());
+        }
+        Node *n = node(slot);
+        free_   = n->next;
+        std::construct_at(&n->key, key);
+        std::construct_at(value(slot), std::forward<Args>(args)...);
+        link_newest(slot);
+        return value(slot);
+    }
+
+    std::optional<V> take(const K &key) noexcept
+    {
+        auto slot = index_.take(key);
+        if (!slot) {
+            return std::nullopt;
+        }
+        return std::optional<V>(release(*slot));
+    }
+
+    bool erase(const K &key) noexcept { return take(key).has_value(); }
+
+    V *oldest() noexcept { return oldest_ == kNone ? nullptr : value(oldest_); }
+    const K *oldest_key() const noexcept { return oldest_ == kNone ? nullptr : &node(oldest_)->key; }
+
+    // Remove the least recently used entry and hand its key and value to
+    // the caller, who decides how to dispose of it.
+    std::optional<std::pair<K, V> > pop_oldest() noexcept
+    {
+        if (oldest_ == kNone) {
+            return std::nullopt;
+        }
+        std::uint32_t slot = oldest_;
+        K key(node(slot)->key);
+        index_.erase(key);
+        V v = release(slot);
+        return std::pair<K, V>(std::move(key), std::move(v));
+    }
+
+    void clear() noexcept
+    {
+        while (oldest_ != kNone) {
+            (void)pop_oldest();
+        }
+    }
+
+    // Visit entries from oldest to newest. The callback may not modify the
+    // container.
+    template <class F>
+    void for_each(F &&f)
+    {
+        for (std::uint32_t s = oldest_; s != kNone; s = node(s)->next) {
+            f(static_cast<const K &>(node(s)->key), *value(s));
+        }
+    }
+
+private:
+    static constexpr std::uint32_t kNone = UINT32_MAX;
+
+    struct Node {
+        union {
+            K key;
+        };
+        alignas(V) std::byte storage[sizeof(V)];
+        std::uint32_t prev = kNone;
+        std::uint32_t next = kNone;
+
+        Node() noexcept {}
+        ~Node() {}
+    };
+
+    Node *node(std::uint32_t slot) const noexcept { return reinterpret_cast<Node *>(nodes_.data()) + slot; }
+    V *value(std::uint32_t slot) const noexcept { return reinterpret_cast<V *>(node(slot)->storage); }
+
+    void link_newest(std::uint32_t slot) noexcept
+    {
+        Node *n = node(slot);
+        n->prev = newest_;
+        n->next = kNone;
+        if (newest_ != kNone) {
+            node(newest_)->next = slot;
+        } else {
+            oldest_ = slot;
+        }
+        newest_ = slot;
+    }
+
+    void unlink(std::uint32_t slot) noexcept
+    {
+        Node *n = node(slot);
+        if (n->prev != kNone) {
+            node(n->prev)->next = n->next;
+        } else {
+            oldest_ = n->next;
+        }
+        if (n->next != kNone) {
+            node(n->next)->prev = n->prev;
+        } else {
+            newest_ = n->prev;
+        }
+        n->prev = n->next = kNone;
+    }
+
+    void touch(std::uint32_t slot) noexcept
+    {
+        if (slot != newest_) {
+            unlink(slot);
+            link_newest(slot);
+        }
+    }
+
+    V release(std::uint32_t slot) noexcept
+    {
+        unlink(slot);
+        Node *n = node(slot);
+        V out(std::move(*value(slot)));
+        std::destroy_at(value(slot));
+        std::destroy_at(&n->key);
+        n->next = free_;
+        free_   = slot;
+        return out;
+    }
+
+    FixedHashMap<K, std::uint32_t, Hash> index_;
+    Region nodes_;
+    std::uint32_t capacity_ = 0;
+    std::uint32_t free_     = kNone;
+    std::uint32_t oldest_   = kNone;
+    std::uint32_t newest_   = kNone;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,137 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_INTRUSIVE_LIST_HPP
+#define SS_CORE_INTRUSIVE_LIST_HPP
+
+// IntrusiveList<T, Tag>: a doubly linked list whose links live inside the
+// elements. Replaces ss_list (core.h).
+//
+// An element type derives from ListNode<Tag> once per list it can be on.
+// The list never owns its elements; it only links objects whose lifetime
+// is managed elsewhere (usually a pool). Destroying a linked node unlinks
+// it, so an element can never leave a dangling link behind.
+//
+// Owner recovery uses static_cast from the base class, which is well
+// defined, instead of ss_container_of pointer arithmetic.
+
+#include <cstddef>
+#include <cstdlib>
+
+namespace ss {
+
+struct DefaultListTag;
+
+template <class Tag = DefaultListTag>
+class ListNode {
+public:
+    ListNode() noexcept = default;
+    ListNode(const ListNode &)            = delete;
+    ListNode &operator=(const ListNode &) = delete;
+    ~ListNode() { unlink(); }
+
+    bool linked() const noexcept { return next_ != nullptr; }
+
+    void unlink() noexcept
+    {
+        if (next_ != nullptr) {
+            prev_->next_ = next_;
+            next_->prev_ = prev_;
+            next_ = prev_ = nullptr;
+        }
+    }
+
+private:
+    template <class, class>
+    friend class IntrusiveList;
+
+    ListNode *prev_ = nullptr;
+    ListNode *next_ = nullptr;
+};
+
+template <class T, class Tag = DefaultListTag>
+class IntrusiveList {
+    using Node = ListNode<Tag>;
+
+public:
+    IntrusiveList() noexcept { head_.prev_ = head_.next_ = &head_; }
+    IntrusiveList(const IntrusiveList &)            = delete;
+    IntrusiveList &operator=(const IntrusiveList &) = delete;
+    ~IntrusiveList() { clear(); }
+
+    bool empty() const noexcept { return head_.next_ == &head_; }
+
+    std::size_t size() const noexcept
+    {
+        std::size_t n = 0;
+        for (const Node *p = head_.next_; p != &head_; p = p->next_) {
+            ++n;
+        }
+        return n;
+    }
+
+    void push_back(T &item) noexcept { insert_before(&head_, node(item)); }
+    void push_front(T &item) noexcept { insert_before(head_.next_, node(item)); }
+
+    // Move an element that is already on this list to the back.
+    void move_to_back(T &item) noexcept
+    {
+        Node *n = node(item);
+        n->unlink();
+        insert_before(&head_, n);
+    }
+
+    static void remove(T &item) noexcept { node(item)->unlink(); }
+
+    T *front() noexcept { return empty() ? nullptr : owner(head_.next_); }
+    T *back() noexcept { return empty() ? nullptr : owner(head_.prev_); }
+
+    T *pop_front() noexcept
+    {
+        T *item = front();
+        if (item != nullptr) {
+            remove(*item);
+        }
+        return item;
+    }
+
+    // Unlink everything without touching the elements' lifetimes.
+    void clear() noexcept
+    {
+        while (!empty()) {
+            head_.next_->unlink();
+        }
+    }
+
+    // Visit every element. The callback may unlink or destroy the element
+    // it was given (the next pointer is read first), but not others.
+    template <class F>
+    void for_each(F &&f)
+    {
+        Node *p = head_.next_;
+        while (p != &head_) {
+            Node *next = p->next_;
+            f(*owner(p));
+            p = next;
+        }
+    }
+
+private:
+    static Node *node(T &item) noexcept { return static_cast<Node *>(&item); }
+    static T *owner(Node *n) noexcept { return static_cast<T *>(n); }
+
+    static void insert_before(Node *pos, Node *n) noexcept
+    {
+        if (n->linked()) {
+            std::abort(); // already on a list using this tag
+        }
+        n->next_     = pos;
+        n->prev_     = pos->prev_;
+        pos->prev_->next_ = n;
+        pos->prev_   = n;
+    }
+
+    Node head_;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,282 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_POOL_HPP
+#define SS_CORE_POOL_HPP
+
+// ObjectPool<T>: fixed-capacity storage for objects of one type.
+//
+// Ownership model
+//   * acquire() constructs a T in a free slot and returns PoolPtr<T>, a
+//     std::unique_ptr whose deleter hands the slot back to the pool. The
+//     PoolPtr is the one and only owner.
+//   * Anything that needs to refer to the object without owning it (a DNS
+//     callback, a timer, the peer of a relay) keeps a Handle<T>. A handle
+//     carries the slot's generation, so resolve() returns nullptr once the
+//     object is gone, even if the slot has been reused since.
+//   * A type that embeds libuv handles cannot be freed while libuv still
+//     points at them. Such types model AsyncClosable: when the owner drops
+//     the PoolPtr, the slot enters the closing state (handles stop
+//     resolving), T::begin_close() starts closing the libuv handles, and the
+//     slot is destroyed and recycled only when T invokes the completion.
+//
+// The pool reserves all of its memory up front, so acquire() and release
+// never call the system allocator. Free slots are poisoned under ASan, which
+// keeps use-after-free detectable even though memory is recycled.
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <memory>
+#include <new>
+#include <utility>
+
+#include "core/region.hpp"
+#include "core/result.hpp"
+
+namespace ss {
+
+template <class T>
+class ObjectPool;
+
+template <class T>
+struct Handle {
+    std::uint32_t index = UINT32_MAX;
+    std::uint32_t gen   = 0;
+
+    constexpr explicit operator bool() const noexcept { return gen != 0; }
+    friend constexpr bool operator==(const Handle &, const Handle &) = default;
+};
+
+// Passed to T::begin_close(). T must invoke it exactly once, after its last
+// asynchronous close has finished; it may also be invoked synchronously from
+// inside begin_close(). The object is destroyed during the call, so T must
+// not touch itself afterwards.
+class CloseCompletion {
+public:
+    CloseCompletion() noexcept = default;
+    CloseCompletion(void (*fn)(void *, std::uint32_t), void *pool, std::uint32_t index) noexcept
+        : fn_(fn), pool_(pool), index_(index)
+    {
+    }
+
+    void operator()() noexcept
+    {
+        auto fn = std::exchange(fn_, nullptr);
+        if (fn == nullptr) {
+            std::abort(); // completed twice, or never armed
+        }
+        fn(pool_, index_);
+    }
+
+    explicit operator bool() const noexcept { return fn_ != nullptr; }
+
+private:
+    void (*fn_)(void *, std::uint32_t) = nullptr;
+    void *pool_                         = nullptr;
+    std::uint32_t index_                = 0;
+};
+
+template <class T>
+concept AsyncClosable = requires(T &t, CloseCompletion done) {
+    t.begin_close(done);
+};
+
+template <class T>
+struct PoolDeleter {
+    ObjectPool<T> *pool = nullptr;
+
+    void operator()(T *object) const noexcept
+    {
+        if (pool == nullptr) {
+            std::abort(); // a PoolPtr must come from ObjectPool::acquire
+        }
+        pool->release(object);
+    }
+};
+
+template <class T>
+using PoolPtr = std::unique_ptr<T, PoolDeleter<T> >;
+
+template <class T>
+class ObjectPool {
+    static_assert(std::is_nothrow_destructible_v<T>);
+
+public:
+    ObjectPool() noexcept = default;
+    ObjectPool(const ObjectPool &)            = delete;
+    ObjectPool &operator=(const ObjectPool &) = delete;
+    ObjectPool(ObjectPool &&)                 = delete;
+    ObjectPool &operator=(ObjectPool &&)      = delete;
+
+    ~ObjectPool()
+    {
+        // Every PoolPtr must be gone and every close finished before the
+        // pool goes away; otherwise a deleter or close callback would write
+        // into freed memory.
+        if (live_ != 0 || closing_ != 0) {
+            std::abort();
+        }
+    }
+
+    // Reserve storage for `capacity` objects. Called once at startup.
+    Status reserve(std::uint32_t capacity) noexcept
+    {
+        if (capacity_ != 0 || capacity == 0 || capacity == UINT32_MAX) {
+            return fail(Errc::invalid);
+        }
+        auto objects = Region::reserve(static_cast<std::size_t>(capacity) * sizeof(T), alignof(T));
+        if (!objects) {
+            return fail(objects.error());
+        }
+        auto meta = Region::reserve(static_cast<std::size_t>(capacity) * sizeof(Meta), alignof(Meta));
+        if (!meta) {
+            return fail(meta.error());
+        }
+        objects_  = std::move(*objects);
+        meta_     = std::move(*meta);
+        capacity_ = capacity;
+        for (std::uint32_t i = 0; i < capacity; ++i) {
+            Meta *m = std::construct_at(meta_at(i));
+            m->next = (i + 1 < capacity) ? i + 1 : kNone;
+        }
+        free_head_ = 0;
+        SS_POISON(objects_.data(), objects_.size());
+        return {};
+    }
+
+    template <class... Args>
+    Result<PoolPtr<T> > acquire(Args &&...args) noexcept
+    {
+        if (free_head_ == kNone) {
+            return fail(Errc::exhausted);
+        }
+        std::uint32_t index = free_head_;
+        Meta *m             = meta_at(index);
+        free_head_          = m->next;
+        m->next             = kNone;
+        m->state            = State::live;
+        ++live_;
+        void *raw = slot_at(index);
+        SS_UNPOISON(raw, sizeof(T));
+        T *object = std::construct_at(static_cast<T *>(raw), std::forward<Args>(args)...);
+        return PoolPtr<T>(object, PoolDeleter<T>{ this });
+    }
+
+    // Handle for a live object owned by this pool.
+    Handle<T> handle_of(const T *object) const noexcept
+    {
+        std::uint32_t index = index_of(object);
+        return Handle<T>{ index, meta_at(index)->gen };
+    }
+
+    // The object a handle refers to, or nullptr once it is closing or gone.
+    T *resolve(Handle<T> handle) const noexcept
+    {
+        if (handle.index >= capacity_) {
+            return nullptr;
+        }
+        const Meta *m = meta_at(handle.index);
+        if (m->state != State::live || m->gen != handle.gen) {
+            return nullptr;
+        }
+        return static_cast<T *>(slot_at(handle.index));
+    }
+
+    bool owns(const T *object) const noexcept
+    {
+        auto *p     = reinterpret_cast<const std::byte *>(object);
+        auto *first = objects_.data();
+        return first != nullptr && p >= first && p < first + objects_.size();
+    }
+
+    std::uint32_t capacity() const noexcept { return capacity_; }
+    std::uint32_t live() const noexcept { return live_; }
+    std::uint32_t closing() const noexcept { return closing_; }
+    std::uint32_t available() const noexcept { return capacity_ - live_ - closing_; }
+    bool full() const noexcept { return free_head_ == kNone; }
+
+private:
+    friend struct PoolDeleter<T>;
+
+    enum class State : std::uint8_t { free, live, closing };
+
+    struct Meta {
+        std::uint32_t gen  = 1; // 0 is reserved for the empty handle
+        std::uint32_t next = kNone;
+        State state        = State::free;
+    };
+
+    static constexpr std::uint32_t kNone = UINT32_MAX;
+
+    void *slot_at(std::uint32_t index) const noexcept
+    {
+        return objects_.data() + static_cast<std::size_t>(index) * sizeof(T);
+    }
+    Meta *meta_at(std::uint32_t index) const noexcept
+    {
+        return reinterpret_cast<Meta *>(meta_.data()) + index;
+    }
+
+    std::uint32_t index_of(const T *object) const noexcept
+    {
+        if (!owns(object)) {
+            std::abort();
+        }
+        auto offset = static_cast<std::size_t>(reinterpret_cast<const std::byte *>(object) - objects_.data());
+        if (offset % sizeof(T) != 0) {
+            std::abort();
+        }
+        return static_cast<std::uint32_t>(offset / sizeof(T));
+    }
+
+    void release(T *object) noexcept
+    {
+        std::uint32_t index = index_of(object);
+        Meta *m             = meta_at(index);
+        if (m->state != State::live) {
+            std::abort(); // double release
+        }
+        --live_;
+        if constexpr (AsyncClosable<T>) {
+            m->state = State::closing;
+            ++closing_;
+            object->begin_close(CloseCompletion(&ObjectPool::closed_trampoline, this, index));
+        } else {
+            recycle(index);
+        }
+    }
+
+    static void closed_trampoline(void *pool, std::uint32_t index) noexcept
+    {
+        auto *self = static_cast<ObjectPool *>(pool);
+        Meta *m    = self->meta_at(index);
+        if (m->state != State::closing) {
+            std::abort();
+        }
+        --self->closing_;
+        self->recycle(index);
+    }
+
+    void recycle(std::uint32_t index) noexcept
+    {
+        Meta *m = meta_at(index);
+        std::destroy_at(static_cast<T *>(slot_at(index)));
+        SS_POISON(slot_at(index), sizeof(T));
+        m->state = State::free;
+        if (++m->gen == 0) {
+            m->gen = 1;
+        }
+        m->next    = free_head_;
+        free_head_ = index;
+    }
+
+    Region objects_;
+    Region meta_;
+    std::uint32_t capacity_  = 0;
+    std::uint32_t free_head_ = kNone;
+    std::uint32_t live_      = 0;
+    std::uint32_t closing_   = 0;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,135 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_REF_HPP
+#define SS_CORE_REF_HPP
+
+// RefCounted<T> / RefPtr<T>: intrusive shared ownership.
+//
+// Use this only where ownership is genuinely shared. Today that means
+// AEAD-2022 UDP sessions, which are held by the session table and by every
+// relay slot that is still talking to the peer. Everything else has a single
+// owner and uses std::unique_ptr (usually PoolPtr).
+//
+// Why not std::shared_ptr: std::allocate_shared has an implementation-
+// defined control block that cannot be sized into a pool portably, weak_ptr
+// would keep the slot alive after the object dies, and the atomic count is
+// wasted on a single-threaded event loop. The count here is a plain integer;
+// RefPtr must not cross threads.
+//
+// T derives from RefCounted<T> and provides
+//     static void destroy(T *object) noexcept;
+// which runs when the last reference goes away (typically by dropping the
+// PoolPtr that owns the storage).
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <utility>
+
+namespace ss {
+
+template <class T>
+class RefPtr;
+
+template <class T>
+class RefCounted {
+public:
+    RefCounted(const RefCounted &)            = delete;
+    RefCounted &operator=(const RefCounted &) = delete;
+
+    std::size_t ref_count() const noexcept { return count_; }
+
+protected:
+    ~RefCounted()
+    {
+        if (count_ != 0) {
+            std::abort(); // destroyed while references remain
+        }
+    }
+
+private:
+    friend T;
+    friend class RefPtr<T>;
+
+    RefCounted() noexcept = default;
+
+    void retain() noexcept
+    {
+        if (count_ == SIZE_MAX) {
+            std::abort();
+        }
+        ++count_;
+    }
+
+    void release() noexcept
+    {
+        if (count_ == 0) {
+            std::abort();
+        }
+        if (--count_ == 0) {
+            T::destroy(static_cast<T *>(this));
+        }
+    }
+
+    std::size_t count_ = 0;
+};
+
+template <class T>
+class RefPtr {
+public:
+    constexpr RefPtr() noexcept = default;
+    constexpr RefPtr(std::nullptr_t) noexcept {} // NOLINT(google-explicit-constructor)
+
+    // Takes the first reference, or adds one to an object that is already
+    // shared.
+    explicit RefPtr(T *object) noexcept : ptr_(object)
+    {
+        if (ptr_ != nullptr) {
+            ptr_->retain();
+        }
+    }
+
+    RefPtr(const RefPtr &other) noexcept : RefPtr(other.ptr_) {}
+    RefPtr(RefPtr &&other) noexcept : ptr_(std::exchange(other.ptr_, nullptr)) {}
+
+    RefPtr &operator=(const RefPtr &other) noexcept
+    {
+        RefPtr(other).swap(*this);
+        return *this;
+    }
+    RefPtr &operator=(RefPtr &&other) noexcept
+    {
+        RefPtr(std::move(other)).swap(*this);
+        return *this;
+    }
+    RefPtr &operator=(std::nullptr_t) noexcept
+    {
+        reset();
+        return *this;
+    }
+
+    ~RefPtr() { reset(); }
+
+    void reset() noexcept
+    {
+        if (T *p = std::exchange(ptr_, nullptr)) {
+            p->release();
+        }
+    }
+
+    void swap(RefPtr &other) noexcept { std::swap(ptr_, other.ptr_); }
+
+    T *get() const noexcept { return ptr_; }
+    T &operator*() const noexcept { return *ptr_; }
+    T *operator->() const noexcept { return ptr_; }
+    explicit operator bool() const noexcept { return ptr_ != nullptr; }
+
+    friend bool operator==(const RefPtr &a, const RefPtr &b) noexcept { return a.ptr_ == b.ptr_; }
+    friend bool operator==(const RefPtr &a, std::nullptr_t) noexcept { return a.ptr_ == nullptr; }
+
+private:
+    T *ptr_ = nullptr;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,117 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_REGION_HPP
+#define SS_CORE_REGION_HPP
+
+// Region: one aligned block of raw memory obtained from the system allocator.
+//
+// This is the only place in src/core that calls malloc. Pools, arenas and
+// tables reserve their storage through a Region while the process starts up;
+// after that the steady-state data path never touches the system allocator.
+
+#include <cstddef>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <utility>
+
+#include "core/result.hpp"
+
+#if defined(__has_feature)
+#if __has_feature(address_sanitizer)
+#define SS_HAVE_ASAN 1
+#endif
+#endif
+#if defined(__SANITIZE_ADDRESS__)
+#define SS_HAVE_ASAN 1
+#endif
+
+#ifdef SS_HAVE_ASAN
+#include <sanitizer/asan_interface.h>
+#define SS_POISON(addr, size)   ASAN_POISON_MEMORY_REGION((addr), (size))
+#define SS_UNPOISON(addr, size) ASAN_UNPOISON_MEMORY_REGION((addr), (size))
+#else
+#define SS_POISON(addr, size)   ((void)(addr), (void)(size))
+#define SS_UNPOISON(addr, size) ((void)(addr), (void)(size))
+#endif
+
+namespace ss {
+
+class Region {
+public:
+    Region() noexcept = default;
+    Region(const Region &)            = delete;
+    Region &operator=(const Region &) = delete;
+    Region(Region &&other) noexcept
+        : base_(std::exchange(other.base_, nullptr))
+        , data_(std::exchange(other.data_, nullptr))
+        , size_(std::exchange(other.size_, 0))
+    {
+    }
+    Region &operator=(Region &&other) noexcept
+    {
+        if (this != &other) {
+            release();
+            base_ = std::exchange(other.base_, nullptr);
+            data_ = std::exchange(other.data_, nullptr);
+            size_ = std::exchange(other.size_, 0);
+        }
+        return *this;
+    }
+    ~Region() { release(); }
+
+    // Reserve `size` zeroed bytes aligned to `align` (a power of two).
+    static Result<Region> reserve(std::size_t size, std::size_t align = alignof(std::max_align_t)) noexcept
+    {
+        if (align == 0 || (align & (align - 1)) != 0) {
+            return fail(Errc::invalid);
+        }
+        if (align < alignof(std::max_align_t)) {
+            align = alignof(std::max_align_t);
+        }
+        if (size == 0) {
+            return Region{};
+        }
+        if (size > SIZE_MAX - align) {
+            return fail(Errc::exhausted);
+        }
+        // Over-allocate and align by hand: aligned_alloc is missing on
+        // MinGW, and _aligned_malloc would need its own free function.
+        void *base = std::calloc(1, size + align); // heap-ok: startup reservation
+        if (base == nullptr) {
+            return fail(Errc::exhausted);
+        }
+        auto addr = reinterpret_cast<std::uintptr_t>(base);
+        auto aligned = (addr + align - 1) & ~(static_cast<std::uintptr_t>(align) - 1);
+        Region r;
+        r.base_ = base;
+        r.data_ = reinterpret_cast<std::byte *>(aligned);
+        r.size_ = size;
+        return r;
+    }
+
+    std::byte *data() const noexcept { return data_; }
+    std::size_t size() const noexcept { return size_; }
+    bool empty() const noexcept { return size_ == 0; }
+
+private:
+    void release() noexcept
+    {
+        if (base_ != nullptr) {
+            // Pools may leave freed slots poisoned; the allocator needs to
+            // be able to reuse the whole block.
+            SS_UNPOISON(data_, size_);
+            std::free(base_); // heap-ok: startup reservation
+        }
+        base_ = nullptr;
+        data_ = nullptr;
+        size_ = 0;
+    }
+
+    void *base_      = nullptr;
+    std::byte *data_ = nullptr;
+    std::size_t size_ = 0;
+};
+
+} // namespace ss
+
+#endif
@@ -0,0 +1,236 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_CORE_RESULT_HPP
+#define SS_CORE_RESULT_HPP
+
+// Error-or-value return type for code built without exceptions.
+//
+// The interface is a subset of C++23 std::expected (has_value, value, error,
+// and_then, transform, value_or) so Result can become an alias of
+// std::expected once the toolchain floor allows it.
+
+#include <cstdint>
+#include <cstdlib>
+#include <memory>
+#include <new>
+#include <type_traits>
+#include <utility>
+
+namespace ss {
+
+enum class Errc : std::uint8_t {
+    exhausted, // a fixed-capacity pool or table is full
+    io,        // socket or file error; Error::sys holds errno
+    again,     // operation would block
+    crypto,    // authentication or cipher failure
+    protocol,  // malformed peer input
+    config,    // invalid configuration
+    closed,    // object is closing or already closed
+    invalid,   // invalid argument from the caller
+};
+
+struct Error {
+    Errc code;
+    int sys = 0;
+
+    friend constexpr bool operator==(const Error &, const Error &) = default;
+};
+
+template <class E>
+struct Unexpected {
+    E value;
+};
+
+template <class E>
+constexpr Unexpected<std::decay_t<E> > fail(E &&e)
+{
+    return { std::forward<E>(e) };
+}
+
+constexpr Unexpected<Error> fail(Errc code, int sys = 0)
+{
+    return { Error{ code, sys } };
+}
+
+namespace detail {
+[[noreturn]] inline void result_misuse() noexcept
+{
+    // Accessing the wrong alternative is a programming error. There is no
+    // exception to throw, so stop here instead of reading garbage.
+    std::abort();
+}
+} // namespace detail
+
+template <class T, class E = Error>
+class [[nodiscard]] Result {
+    static_assert(!std::is_reference_v<T>, "Result<T&> is not supported; use a pointer");
+
+public:
+    using value_type = T;
+    using error_type = E;
+
+    template <class U = T>
+    requires std::is_constructible_v<T, U &&>
+    constexpr Result(U &&v) noexcept(std::is_nothrow_constructible_v<T, U &&>) // NOLINT(google-explicit-constructor)
+        : ok_(true)
+    {
+        std::construct_at(&value_, std::forward<U>(v));
+    }
+
+    template <class G>
+    constexpr Result(Unexpected<G> &&e) noexcept // NOLINT(google-explicit-constructor)
+        : ok_(false)
+    {
+        std::construct_at(&error_, std::move(e.value));
+    }
+
+    Result(const Result &other) requires std::is_copy_constructible_v<T>
+        : ok_(other.ok_)
+    {
+        if (ok_) {
+            std::construct_at(&value_, other.value_);
+        } else {
+            std::construct_at(&error_, other.error_);
+        }
+    }
+
+    Result(Result &&other) noexcept(std::is_nothrow_move_constructible_v<T>)
+        : ok_(other.ok_)
+    {
+        if (ok_) {
+            std::construct_at(&value_, std::move(other.value_));
+        } else {
+            std::construct_at(&error_, std::move(other.error_));
+        }
+    }
+
+    Result &operator=(const Result &) = delete;
+    Result &operator=(Result &&)      = delete;
+
+    ~Result()
+    {
+        if (ok_) {
+            std::destroy_at(&value_);
+        } else {
+            std::destroy_at(&error_);
+        }
+    }
+
+    constexpr bool has_value() const noexcept { return ok_; }
+    constexpr explicit operator bool() const noexcept { return ok_; }
+
+    constexpr T &value() & noexcept
+    {
+        if (!ok_) {
+            detail::result_misuse();
+        }
+        return value_;
+    }
+    constexpr const T &value() const & noexcept
+    {
+        if (!ok_) {
+            detail::result_misuse();
+        }
+        return value_;
+    }
+    constexpr T &&value() && noexcept
+    {
+        if (!ok_) {
+            detail::result_misuse();
+        }
+        return std::move(value_);
+    }
+
+    constexpr T &operator*() & noexcept { return value(); }
+    constexpr const T &operator*() const & noexcept { return value(); }
+    constexpr T &&operator*() && noexcept { return std::move(*this).value(); }
+    constexpr T *operator->() noexcept { return &value(); }
+    constexpr const T *operator->() const noexcept { return &value(); }
+
+    constexpr const E &error() const noexcept
+    {
+        if (ok_) {
+            detail::result_misuse();
+        }
+        return error_;
+    }
+
+    template <class U>
+    constexpr T value_or(U &&fallback) const & { return ok_ ? value_ : static_cast<T>(std::forward<U>(fallback)); }
+
+    template <class F>
+    constexpr auto and_then(F &&f) &&
+    {
+        using R = std::invoke_result_t<F, T &&>;
+        if (ok_) {
+            return std::forward<F>(f)(std::move(value_));
+        }
+        return R(Unexpected<E>{ error_ });
+    }
+
+    template <class F>
+    constexpr auto transform(F &&f) &&
+    {
+        using U = std::invoke_result_t<F, T &&>;
+        if (ok_) {
+            return Result<U, E>(std::forward<F>(f)(std::move(value_)));
+        }
+        return Result<U, E>(Unexpected<E>{ error_ });
+    }
+
+private:
+    union {
+        T value_;
+        E error_;
+    };
+    bool ok_;
+};
+
+template <class E>
+class [[nodiscard]] Result<void, E> {
+public:
+    using value_type = void;
+    using error_type = E;
+
+    constexpr Result() noexcept : ok_(true), error_() {}
+
+    template <class G>
+    constexpr Result(Unexpected<G> &&e) noexcept // NOLINT(google-explicit-constructor)
+        : ok_(false), error_(std::move(e.value))
+    {
+    }
+
+    constexpr bool has_value() const noexcept { return ok_; }
+    constexpr explicit operator bool() const noexcept { return ok_; }
+    constexpr void value() const noexcept
+    {
+        if (!ok_) {
+            detail::result_misuse();
+        }
+    }
+    constexpr const E &error() const noexcept
+    {
+        if (ok_) {
+            detail::result_misuse();
+        }
+        return error_;
+    }
+
+private:
+    bool ok_;
+    E error_;
+};
+
+using Status = Result<void>;
+
+// Propagate an error out of a function returning any Result.
+#define SS_TRY(expr)                                         \
+    do {                                                     \
+        auto ss_try_result_ = (expr);                        \
+        if (!ss_try_result_) {                               \
+            return ::ss::fail(ss_try_result_.error());       \
+        }                                                    \
+    } while (0)
+
+} // namespace ss
+
+#endif
@@ -2588,6 +2588,12 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`.
        }
    }

+#ifdef __MINGW32__
+    if (manager_addr != NULL) {
+        LOGE("--manager-address is not supported on Windows; ignored");
+    }
+#endif
+
 #ifndef __MINGW32__
    if (manager_addr != NULL) {
        ss_timer_init(&stat_update_watcher, stat_update_cb, UPDATE_INTERVAL, UPDATE_INTERVAL);
@@ -57,6 +57,19 @@ function(ss_add_test name sources libs)
    endif()
 endfunction()

+# C++ core (src/core): header-only ownership and allocation primitives.
+foreach(core_test result pool buffer arena fixed_map ownership)
+    ss_add_test(test_core_${core_test} "test_core_${core_test}.cpp" "")
+endforeach()
+
+# C++ sources under src/ must not allocate on the heap (see the script).
+find_package(Python3 COMPONENTS Interpreter QUIET)
+if(Python3_Interpreter_FOUND)
+    add_test(NAME check_no_heap
+        COMMAND ${Python3_EXECUTABLE} ${PROJECT_SOURCE_DIR}/scripts/check_no_heap.py --self-test)
+    set_tests_properties(check_no_heap PROPERTIES LABELS "unit")
+endif()
+
 # test_base64 - no external deps
 ss_add_test(test_base64
    "test_base64.c;${PROJECT_SOURCE_DIR}/src/base64.c"
@@ -215,6 +228,15 @@ ss_add_test(test_resolv_async
    "ss_ev;ss_cares;ss_sodium")
 set_tests_properties(test_resolv_async PROPERTIES TIMEOUT 30)

+# Preloadable malloc counter for the allocation audit. It interposes the
+# system allocator, so it is only built where that is supported (glibc,
+# macOS) and never together with the sanitizers, which own malloc.
+include(CheckSymbolExists)
+check_symbol_exists(__GLIBC__ "stdlib.h" SS_HAVE_GLIBC)
+if(SS_BUILD_EXECUTABLES AND NOT ENABLE_SANITIZERS AND (APPLE OR SS_HAVE_GLIBC))
+    add_library(ss_alloc_audit SHARED alloc_audit/alloc_counter.c)
+endif()
+
 if(SS_BUILD_EXECUTABLES AND NOT CMAKE_CROSSCOMPILING)
    find_package(Python3 COMPONENTS Interpreter QUIET)
    if(Python3_Interpreter_FOUND)
@@ -233,6 +255,15 @@ if(SS_BUILD_EXECUTABLES AND NOT CMAKE_CROSSCOMPILING)
            COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/experiments/test_transport.py)
        set_tests_properties(test_experimental_transports PROPERTIES
            LABELS "integration" TIMEOUT 30)
+        # Steady-state allocation audit (report-only until the C++ relay
+        # lands; see tests/alloc_audit/measure_alloc.py).
+        if(TARGET ss_alloc_audit)
+            add_test(NAME alloc_audit_report
+                COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/alloc_audit/measure_alloc.py
+                        --bin ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}
+                        --shim $<TARGET_FILE:ss_alloc_audit>)
+            set_tests_properties(alloc_audit_report PROPERTIES LABELS "integration" TIMEOUT 120)
+        endif()
        add_test(NAME test_printable_salt_interop
            COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/interop.py
                    --self --printable-salt --bin ${CMAKE_RUNTIME_OUTPUT_DIRECTORY})
@@ -0,0 +1,217 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+/*
+ * alloc_counter: a preloadable shim that counts heap allocations.
+ *
+ * Load it into a program with LD_PRELOAD (glibc) or DYLD_INSERT_LIBRARIES
+ * (macOS). Every SIGUSR2 appends one line with the running totals to the
+ * file named by SS_ALLOC_AUDIT_OUT:
+ *
+ *     <pid> <allocations> <frees> <bytes>
+ *
+ * A driver takes a snapshot after warming a proxy up and another after more
+ * traffic; the difference is what the steady-state data path allocates.
+ * tests/alloc_audit/measure_alloc.py is that driver.
+ *
+ * The signal handler only uses async-signal-safe calls (open, write, close).
+ */
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
+#include <errno.h>
+#include <fcntl.h>
+#include <signal.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+static uint64_t allocations;
+static uint64_t frees;
+static uint64_t bytes;
+static const char *out_path;
+
+static void
+count_alloc(size_t size)
+{
+    __atomic_add_fetch(&allocations, 1, __ATOMIC_RELAXED);
+    __atomic_add_fetch(&bytes, (uint64_t)size, __ATOMIC_RELAXED);
+}
+
+static void
+count_free(void *ptr)
+{
+    if (ptr != NULL) {
+        __atomic_add_fetch(&frees, 1, __ATOMIC_RELAXED);
+    }
+}
+
+static char *
+format_u64(char *end, uint64_t value)
+{
+    do {
+        *--end = (char)('0' + value % 10);
+        value /= 10;
+    } while (value != 0);
+    return end;
+}
+
+static void
+snapshot(int signo)
+{
+    (void)signo;
+    int saved_errno = errno;
+    if (out_path != NULL) {
+        uint64_t fields[4] = {
+            (uint64_t)getpid(),
+            __atomic_load_n(&allocations, __ATOMIC_RELAXED),
+            __atomic_load_n(&frees, __ATOMIC_RELAXED),
+            __atomic_load_n(&bytes, __ATOMIC_RELAXED),
+        };
+        char line[128];
+        char *end = line + sizeof(line);
+        char *p   = end;
+        *--p = '\n';
+        for (int i = 3; i >= 0; i--) {
+            p = format_u64(p, fields[i]);
+            if (i != 0) {
+                *--p = ' ';
+            }
+        }
+        int fd = open(out_path, O_WRONLY | O_CREAT | O_APPEND, 0644);
+        if (fd >= 0) {
+            ssize_t ignored = write(fd, p, (size_t)(end - p));
+            (void)ignored;
+            close(fd);
+        }
+    }
+    errno = saved_errno;
+}
+
+__attribute__((constructor)) static void
+install(void)
+{
+    out_path = getenv("SS_ALLOC_AUDIT_OUT");
+    struct sigaction action;
+    memset(&action, 0, sizeof(action));
+    action.sa_handler = snapshot;
+    action.sa_flags   = SA_RESTART;
+    sigemptyset(&action.sa_mask);
+    sigaction(SIGUSR2, &action, NULL);
+}
+
+#if defined(__APPLE__)
+
+#include <malloc/malloc.h>
+
+static void *
+audit_malloc(size_t size)
+{
+    count_alloc(size);
+    return malloc(size);
+}
+
+static void *
+audit_calloc(size_t count, size_t size)
+{
+    count_alloc(count * size);
+    return calloc(count, size);
+}
+
+static void *
+audit_realloc(void *ptr, size_t size)
+{
+    count_alloc(size);
+    return realloc(ptr, size);
+}
+
+static int
+audit_posix_memalign(void **out, size_t align, size_t size)
+{
+    count_alloc(size);
+    return posix_memalign(out, align, size);
+}
+
+static void
+audit_free(void *ptr)
+{
+    count_free(ptr);
+    free(ptr);
+}
+
+#define SS_INTERPOSE(replacement, original)                                   \
+    __attribute__((used)) static const struct {                               \
+        const void *replacement_fn;                                           \
+        const void *original_fn;                                              \
+    } interpose_##original __attribute__((section("__DATA,__interpose"))) = { \
+        (const void *)(uintptr_t)&replacement, (const void *)(uintptr_t)&original }
+
+SS_INTERPOSE(audit_malloc, malloc);
+SS_INTERPOSE(audit_calloc, calloc);
+SS_INTERPOSE(audit_realloc, realloc);
+SS_INTERPOSE(audit_posix_memalign, posix_memalign);
+SS_INTERPOSE(audit_free, free);
+
+#elif defined(__GLIBC__)
+
+extern void *__libc_malloc(size_t);
+extern void *__libc_calloc(size_t, size_t);
+extern void *__libc_realloc(void *, size_t);
+extern void *__libc_memalign(size_t, size_t);
+extern void __libc_free(void *);
+
+void *
+malloc(size_t size)
+{
+    count_alloc(size);
+    return __libc_malloc(size);
+}
+
+void *
+calloc(size_t count, size_t size)
+{
+    count_alloc(count * size);
+    return __libc_calloc(count, size);
+}
+
+void *
+realloc(void *ptr, size_t size)
+{
+    count_alloc(size);
+    return __libc_realloc(ptr, size);
+}
+
+void *
+memalign(size_t align, size_t size)
+{
+    count_alloc(size);
+    return __libc_memalign(align, size);
+}
+
+void *
+aligned_alloc(size_t align, size_t size)
+{
+    return memalign(align, size);
+}
+
+int
+posix_memalign(void **out, size_t align, size_t size)
+{
+    void *p = memalign(align, size);
+    if (p == NULL) {
+        return ENOMEM;
+    }
+    *out = p;
+    return 0;
+}
+
+void
+free(void *ptr)
+{
+    count_free(ptr);
+    __libc_free(ptr);
+}
+
+#else
+#error "alloc_counter supports glibc and macOS only"
+#endif
@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Count heap allocations per proxied TCP connection in ss-local and ss-server.
+
+Both peers run with the alloc_counter shim preloaded. After a warm-up the
+script snapshots each process's allocation counter, relays more connections,
+and snapshots again. The difference, divided by the number of connections,
+is what one connection costs in the steady state. The no-heap goal for the
+C++ core is zero.
+
+Report-only by default; --max-per-connection N turns it into a check.
+
+Example:
+    python3 tests/alloc_audit/measure_alloc.py --bin build/bin \\
+        --shim build/lib/libss_alloc_audit.dylib --method aes-256-gcm
+"""
+import argparse
+import base64
+import concurrent.futures
+import json
+import os
+from pathlib import Path
+import signal
+import sys
+import tempfile
+import threading
+import time
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from interop import TCPHandler, TCPOrigin, free_port, peers, tcp_case  # noqa: E402
+
+
+def read_snapshot(path, previous_lines, timeout=10):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        lines = path.read_text().splitlines() if path.exists() else []
+        if len(lines) > previous_lines:
+            pid, allocations, frees, size = (int(v) for v in lines[-1].split())
+            return {"pid": pid, "allocations": allocations, "frees": frees, "bytes": size}, len(lines)
+        time.sleep(.02)
+    raise AssertionError(f"no allocation snapshot written to {path}")
+
+
+def snapshot_all(processes, outputs, counts):
+    results = []
+    for index, (process, path) in enumerate(zip(processes, outputs)):
+        os.kill(process.pid, signal.SIGUSR2)
+        result, counts[index] = read_snapshot(path, counts[index])
+        results.append(result)
+    return results
+
+
+def relay(local_port, origin_port, connections, concurrency, size):
+    with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as pool:
+        futures = [pool.submit(tcp_case, local_port, origin_port, size) for _ in range(connections)]
+        for future in futures:
+            future.result(timeout=60)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--bin", default="build/bin", help="directory with ss-local and ss-server")
+    parser.add_argument("--shim", required=True, help="path to the alloc_counter shared library")
+    parser.add_argument("--method", default="aes-256-gcm")
+    parser.add_argument("--warmup", type=int, default=50, help="connections before the first snapshot")
+    parser.add_argument("--connections", type=int, default=200, help="connections between snapshots")
+    parser.add_argument("--concurrency", type=int, default=10)
+    parser.add_argument("--size", type=int, default=16384, help="bytes echoed per connection")
+    parser.add_argument("--max-per-connection", type=float, help="fail if any peer exceeds this")
+    parser.add_argument("--json", action="store_true", help="print the result as JSON")
+    args = parser.parse_args()
+
+    binary_dir = Path(args.bin).resolve()
+    shim = str(Path(args.shim).resolve())
+    preload = "DYLD_INSERT_LIBRARIES" if sys.platform == "darwin" else "LD_PRELOAD"
+    key_size = 16 if args.method == "2022-blake3-aes-128-gcm" else 32
+    password = base64.b64encode(os.urandom(key_size)).decode("ascii")
+
+    with TCPOrigin(("127.0.0.1", 0), TCPHandler) as origin, \
+            tempfile.TemporaryDirectory(prefix="ss-alloc-") as temp:
+        threading.Thread(target=origin.serve_forever, daemon=True).start()
+        server_port, local_port = free_port(), free_port()
+        names = ["ss-server", "ss-local"]
+        outputs = [Path(temp) / f"{name}.counts" for name in names]
+        commands = [
+            [str(binary_dir / "ss-server"), "-s", "127.0.0.1", "-p", str(server_port),
+             "-k", password, "-m", args.method],
+            [str(binary_dir / "ss-local"), "-s", "127.0.0.1", "-p", str(server_port),
+             "-l", str(local_port), "-k", password, "-m", args.method],
+        ]
+        environments = []
+        for output in outputs:
+            env = os.environ.copy()
+            env[preload] = shim
+            env["SS_ALLOC_AUDIT_OUT"] = str(output)
+            environments.append(env)
+
+        # peers() takes one environment for all of its commands, so each
+        # peer gets its own context and its own output file.
+        with peers([commands[0]], [server_port], environments[0]) as server, \
+                peers([commands[1]], [local_port], environments[1]) as local:
+            processes = server + local
+            counts = [0, 0]
+            relay(local_port, origin.server_address[1], args.warmup, args.concurrency, args.size)
+            time.sleep(.5)
+            before = snapshot_all(processes, outputs, counts)
+            relay(local_port, origin.server_address[1], args.connections, args.concurrency, args.size)
+            time.sleep(.5)
+            after = snapshot_all(processes, outputs, counts)
+
+    report = {"method": args.method, "connections": args.connections, "size": args.size, "peers": {}}
+    failed = False
+    for name, b, a in zip(names, before, after):
+        delta = a["allocations"] - b["allocations"]
+        per = delta / args.connections
+        report["peers"][name] = {
+            "allocations": delta,
+            "frees": a["frees"] - b["frees"],
+            "bytes": a["bytes"] - b["bytes"],
+            "allocations_per_connection": round(per, 2),
+        }
+        if args.max_per_connection is not None and per > args.max_per_connection:
+            failed = True
+
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        print(f"method {args.method}, {args.connections} connections of {args.size} bytes")
+        for name, peer in report["peers"].items():
+            print(f"  {name:9} {peer['allocations_per_connection']:8.2f} allocations/connection "
+                  f"({peer['allocations']} total, {peer['bytes']} bytes)")
+    if failed:
+        print(f"FAIL: more than {args.max_per_connection} allocations per connection")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -159,7 +159,7 @@ def peers(commands, ports, environment=None):
                process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT, env=environment)
                processes.append(process)
                wait_ready(process, port)
-            yield
+            yield processes
        except BaseException:
            for path in logs:
                print(path.name + ":\n" + path.read_text(errors="replace")[-6000:], flush=True)
@@ -0,0 +1,63 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#include <cstdint>
+#include <cstring>
+
+#include "core/arena.hpp"
+#include "test_helpers.hpp"
+
+namespace {
+
+struct Pair {
+    int a;
+    double b;
+};
+
+void test_allocate_alignment()
+{
+    ss::MonotonicArena arena;
+    SS_CHECK(arena.reserve(256));
+    auto c = arena.allocate(1, 1);
+    SS_CHECK(c);
+    auto d = arena.allocate(8, 64);
+    SS_CHECK(d);
+    SS_CHECK(reinterpret_cast<std::uintptr_t>(d->data()) % 64 == 0);
+    auto p = arena.make<Pair>(Pair{ 3, 1.5 });
+    SS_CHECK(p && (*p)->a == 3 && (*p)->b == 1.5);
+    SS_CHECK(!arena.allocate(8, 3));
+}
+
+void test_exhaustion_and_reset()
+{
+    ss::MonotonicArena arena;
+    SS_CHECK(!arena.allocate(1));
+    SS_CHECK(arena.reserve(64));
+    SS_CHECK(arena.allocate(64, 1));
+    auto over = arena.allocate(1, 1);
+    SS_CHECK(!over && over.error().code == ss::Errc::exhausted);
+    arena.reset();
+    SS_CHECK(arena.used() == 0 && arena.peak() == 64);
+    SS_CHECK(arena.allocate(32, 1));
+}
+
+void test_intern_and_arrays()
+{
+    ss::MonotonicArena arena;
+    SS_CHECK(arena.reserve(128));
+    auto s = arena.intern("aes-256-gcm");
+    SS_CHECK(s && *s == "aes-256-gcm");
+    SS_CHECK(s->data()[s->size()] == '\0');
+    auto empty = arena.intern("");
+    SS_CHECK(empty && empty->empty());
+    auto ports = arena.make_array<std::uint16_t>(4);
+    SS_CHECK(ports && ports->size() == 4 && (*ports)[3] == 0);
+}
+
+} // namespace
+
+int main()
+{
+    SS_RUN(test_allocate_alignment);
+    SS_RUN(test_exhaustion_and_reset);
+    SS_RUN(test_intern_and_arrays);
+    return 0;
+}
@@ -0,0 +1,82 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#include <cstring>
+#include <string_view>
+
+#include "core/buffer.hpp"
+#include "test_helpers.hpp"
+
+namespace {
+
+ss::ConstBytes bytes(std::string_view s)
+{
+    return ss::as_bytes(s.data(), s.size());
+}
+
+bool equals(ss::ConstBytes b, std::string_view s)
+{
+    return b.size() == s.size() && std::memcmp(b.data(), s.data(), s.size()) == 0;
+}
+
+void test_append_consume()
+{
+    ss::FixedBuffer<8> buf;
+    SS_CHECK(buf.empty() && buf.capacity() == 8);
+    SS_CHECK(buf.append(bytes("hello")));
+    SS_CHECK(equals(buf.readable(), "hello"));
+    SS_CHECK(buf.consume(2));
+    SS_CHECK(equals(buf.readable(), "llo"));
+    SS_CHECK(!buf.consume(4));
+    // Appending past the tail compacts first.
+    SS_CHECK(buf.append(bytes("world")));
+    SS_CHECK(equals(buf.readable(), "lloworld"));
+    SS_CHECK(!buf.append(bytes("x")));
+    SS_CHECK(buf.consume(8) && buf.empty());
+}
+
+void test_commit_writable()
+{
+    ss::FixedBuffer<4> buf;
+    auto w = buf.writable();
+    SS_CHECK(w.size() == 4);
+    std::memcpy(w.data(), "abcd", 4);
+    SS_CHECK(buf.commit(4));
+    SS_CHECK(buf.full() && buf.writable().empty());
+    SS_CHECK(!buf.commit(1));
+    SS_CHECK(equals(buf.readable(), "abcd"));
+}
+
+void test_prepend()
+{
+    ss::FixedBuffer<10> buf;
+    SS_CHECK(buf.append(bytes("data")));
+    // No headroom: payload shifts right.
+    SS_CHECK(buf.prepend(bytes("hdr")));
+    SS_CHECK(equals(buf.readable(), "hdrdata"));
+    SS_CHECK(buf.consume(3));
+    // Headroom from the consumed bytes is reused in place.
+    SS_CHECK(buf.prepend(bytes("ab")));
+    SS_CHECK(equals(buf.readable(), "abdata"));
+    SS_CHECK(!buf.prepend(bytes("12345")));
+    SS_CHECK(equals(buf.readable(), "abdata"));
+}
+
+void test_assign_wipe()
+{
+    ss::FixedBuffer<4> buf;
+    SS_CHECK(!buf.assign(bytes("toolong")));
+    SS_CHECK(buf.assign(bytes("key")));
+    SS_CHECK(equals(buf.readable(), "key"));
+    buf.wipe();
+    SS_CHECK(buf.empty());
+}
+
+} // namespace
+
+int main()
+{
+    SS_RUN(test_append_consume);
+    SS_RUN(test_commit_writable);
+    SS_RUN(test_prepend);
+    SS_RUN(test_assign_wipe);
+    return 0;
+}
@@ -0,0 +1,148 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <string_view>
+
+#include "core/fixed_map.hpp"
+#include "test_helpers.hpp"
+
+namespace {
+
+using Key = ss::FixedKey<16>;
+
+Key key(std::string_view s)
+{
+    auto k = Key::from(ss::as_bytes(s.data(), s.size()));
+    SS_CHECK(k.has_value());
+    return *k;
+}
+
+// Every key lands in the same home slot, to exercise probe runs and the
+// backward-shift deletion.
+struct CollidingHash {
+    std::uint64_t operator()(std::uint32_t, std::uint64_t) const noexcept { return 7; }
+};
+
+void test_map_basic()
+{
+    ss::FixedHashMap<std::uint32_t, int> map;
+    SS_CHECK(map.reserve(3, 0x1234));
+    SS_CHECK(map.emplace(1, 10));
+    SS_CHECK(map.emplace(2, 20));
+    SS_CHECK(map.emplace(3, 30));
+    auto dup = map.emplace(2, 99);
+    SS_CHECK(!dup && dup.error().code == ss::Errc::invalid);
+    auto over = map.emplace(4, 40);
+    SS_CHECK(!over && over.error().code == ss::Errc::exhausted);
+    SS_CHECK(*map.find(2) == 20);
+    auto taken = map.take(2);
+    SS_CHECK(taken && *taken == 20);
+    SS_CHECK(!map.find(2) && map.size() == 2);
+    SS_CHECK(map.emplace(4, 40));
+    SS_CHECK(!map.erase(5));
+}
+
+void test_map_backward_shift()
+{
+    ss::FixedHashMap<std::uint32_t, int, CollidingHash> map;
+    SS_CHECK(map.reserve(8));
+    for (std::uint32_t i = 0; i < 8; ++i) {
+        SS_CHECK(map.emplace(i, static_cast<int>(i) * 100));
+    }
+    // Remove from the middle of the probe run; later entries must still be
+    // reachable without tombstones.
+    SS_CHECK(map.erase(3));
+    SS_CHECK(map.erase(0));
+    for (std::uint32_t i = 0; i < 8; ++i) {
+        const int *v = map.find(i);
+        if (i == 0 || i == 3) {
+            SS_CHECK(v == nullptr);
+        } else {
+            SS_CHECK(v != nullptr && *v == static_cast<int>(i) * 100);
+        }
+    }
+    int count = 0;
+    map.for_each([&](std::uint32_t, int &) { ++count; });
+    SS_CHECK(count == 6);
+}
+
+void test_map_move_only_values()
+{
+    ss::FixedHashMap<std::uint32_t, std::unique_ptr<int> > map;
+    SS_CHECK(map.reserve(2));
+    SS_CHECK(map.emplace(1, std::make_unique<int>(5)));
+    auto v = map.take(1);
+    SS_CHECK(v && **v == 5);
+}
+
+void test_lru_order()
+{
+    ss::FixedLru<Key, int> lru;
+    SS_CHECK(lru.reserve(3, 42));
+    SS_CHECK(lru.emplace(key("a"), 1));
+    SS_CHECK(lru.emplace(key("b"), 2));
+    SS_CHECK(lru.emplace(key("c"), 3));
+    SS_CHECK(lru.full());
+    auto over = lru.emplace(key("d"), 4);
+    SS_CHECK(!over && over.error().code == ss::Errc::exhausted);
+
+    // Touch "a" so "b" becomes the oldest.
+    SS_CHECK(*lru.find(key("a")) == 1);
+    SS_CHECK(*lru.oldest_key() == key("b"));
+    // peek() does not change the order.
+    SS_CHECK(*lru.peek(key("b")) == 2);
+    SS_CHECK(*lru.oldest_key() == key("b"));
+
+    auto evicted = lru.pop_oldest();
+    SS_CHECK(evicted && evicted->first == key("b") && evicted->second == 2);
+    SS_CHECK(lru.emplace(key("d"), 4));
+
+    int order[3] = {};
+    int n        = 0;
+    lru.for_each([&](const Key &, int &v) { order[n++] = v; });
+    SS_CHECK(n == 3 && order[0] == 3 && order[1] == 1 && order[2] == 4);
+}
+
+void test_lru_stable_values()
+{
+    ss::FixedLru<std::uint32_t, int> lru;
+    SS_CHECK(lru.reserve(64));
+    auto first = lru.emplace(0, 1000);
+    SS_CHECK(first);
+    int *stable = *first;
+    for (std::uint32_t i = 1; i < 64; ++i) {
+        SS_CHECK(lru.emplace(i, static_cast<int>(i)));
+    }
+    for (std::uint32_t i = 1; i < 64; i += 2) {
+        SS_CHECK(lru.erase(i));
+    }
+    // Erasing others reshuffles the index but never moves values.
+    SS_CHECK(lru.peek(0) == stable && *stable == 1000);
+    auto taken = lru.take(0);
+    SS_CHECK(taken && *taken == 1000);
+    SS_CHECK(lru.size() == 31);
+    lru.clear();
+    SS_CHECK(lru.empty() && lru.oldest() == nullptr);
+}
+
+void test_key_limits()
+{
+    char big[17] = {};
+    SS_CHECK(!Key::from(ss::as_bytes(big, sizeof(big))));
+    SS_CHECK(key("ab") != key("abc"));
+    SS_CHECK(key("") == key(""));
+}
+
+} // namespace
+
+int main()
+{
+    SS_RUN(test_map_basic);
+    SS_RUN(test_map_backward_shift);
+    SS_RUN(test_map_move_only_values);
+    SS_RUN(test_lru_order);
+    SS_RUN(test_lru_stable_values);
+    SS_RUN(test_key_limits);
+    return 0;
+}
@@ -0,0 +1,145 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+// IntrusiveList, RefPtr and FixedHeap: the remaining ownership primitives.
+#include <cstdint>
+#include <cstring>
+#include <utility>
+
+#include "core/fixed_heap.hpp"
+#include "core/intrusive_list.hpp"
+#include "core/pool.hpp"
+#include "core/ref.hpp"
+#include "test_helpers.hpp"
+
+namespace {
+
+struct IdleTag;
+
+struct Conn : ss::ListNode<>, ss::ListNode<IdleTag> {
+    explicit Conn(int v) noexcept : id(v) {}
+    int id;
+};
+
+void test_list_basic()
+{
+    ss::IntrusiveList<Conn> all;
+    ss::IntrusiveList<Conn, IdleTag> idle;
+    Conn a(1), b(2), c(3);
+    all.push_back(a);
+    all.push_back(b);
+    all.push_front(c);
+    idle.push_back(b);
+    SS_CHECK(all.size() == 3 && idle.size() == 1);
+    SS_CHECK(all.front()->id == 3 && all.back()->id == 2);
+
+    all.move_to_back(c);
+    SS_CHECK(all.back()->id == 3);
+
+    // Being on one list does not affect the other.
+    ss::IntrusiveList<Conn>::remove(b);
+    SS_CHECK(all.size() == 2 && idle.front() == &b);
+
+    int sum = 0;
+    all.for_each([&](Conn &x) { sum += x.id; });
+    SS_CHECK(sum == 4);
+    SS_CHECK(all.pop_front()->id == 1);
+}
+
+void test_list_unlinks_on_destruction()
+{
+    ss::IntrusiveList<Conn> all;
+    {
+        Conn temp(9);
+        all.push_back(temp);
+        SS_CHECK(!all.empty());
+    }
+    SS_CHECK(all.empty());
+}
+
+// A pooled, shared object: the pool owns the storage, references own the
+// object's lifetime.
+struct Session : ss::RefCounted<Session> {
+    static inline ss::ObjectPool<Session> *pool = nullptr;
+    static inline int destroyed                 = 0;
+
+    // The PoolPtr is parked inside the object while references exist.
+    ss::PoolPtr<Session> self;
+
+    static void destroy(Session *s) noexcept
+    {
+        ++destroyed;
+        ss::PoolPtr<Session> owner = std::move(s->self);
+        owner.reset(); // returns the slot to the pool
+    }
+};
+
+ss::RefPtr<Session> make_session()
+{
+    auto slot = Session::pool->acquire();
+    SS_CHECK(slot);
+    Session *raw = slot->get();
+    raw->self    = std::move(slot).value();
+    return ss::RefPtr<Session>(raw);
+}
+
+void test_refptr_lifecycle()
+{
+    ss::ObjectPool<Session> pool;
+    SS_CHECK(pool.reserve(2));
+    Session::pool      = &pool;
+    Session::destroyed = 0;
+    {
+        ss::RefPtr<Session> table = make_session();
+        SS_CHECK(table->ref_count() == 1);
+        ss::RefPtr<Session> relay = table;
+        SS_CHECK(table->ref_count() == 2);
+        ss::RefPtr<Session> moved = std::move(relay);
+        SS_CHECK(!relay && moved->ref_count() == 2); // NOLINT(bugprone-use-after-move): checks moved-from state
+        table.reset();
+        SS_CHECK(Session::destroyed == 0 && pool.live() == 1);
+        moved = nullptr;
+        SS_CHECK(Session::destroyed == 1 && pool.live() == 0);
+    }
+    SS_CHECK(pool.available() == 2);
+}
+
+void test_fixed_heap()
+{
+    ss::FixedHeap heap;
+    SS_CHECK(heap.reserve(4096));
+    void *a = heap.allocate(10);
+    void *b = heap.allocate_zeroed(4, 25);
+    SS_CHECK(a && b && heap.owns(a) && heap.owns(b));
+    SS_CHECK(reinterpret_cast<std::uintptr_t>(a) % 16 == 0);
+    auto *zeros = static_cast<unsigned char *>(b);
+    for (int i = 0; i < 100; ++i) {
+        SS_CHECK(zeros[i] == 0);
+    }
+    std::memcpy(a, "0123456789", 10);
+    void *grown = heap.reallocate(a, 200);
+    SS_CHECK(grown && std::memcmp(grown, "0123456789", 10) == 0);
+
+    auto before = heap.stats();
+    heap.deallocate(grown);
+    heap.deallocate(b);
+    SS_CHECK(heap.stats().in_use < before.in_use);
+
+    // Freed blocks are reused instead of consuming fresh space.
+    void *again = heap.allocate(200);
+    SS_CHECK(again == grown);
+    heap.deallocate(again);
+
+    SS_CHECK(heap.allocate(ss::FixedHeap::max_request() + 1) == nullptr);
+    SS_CHECK(heap.allocate(8192) == nullptr); // over budget
+    SS_CHECK(heap.stats().failures == 2);
+}
+
+} // namespace
+
+int main()
+{
+    SS_RUN(test_list_basic);
+    SS_RUN(test_list_unlinks_on_destruction);
+    SS_RUN(test_refptr_lifecycle);
+    SS_RUN(test_fixed_heap);
+    return 0;
+}
@@ -0,0 +1,140 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#include <cstdint>
+#include <utility>
+
+#include "core/pool.hpp"
+#include "test_helpers.hpp"
+
+namespace {
+
+int g_destroyed = 0;
+
+struct Plain {
+    explicit Plain(int v) noexcept : value(v) {}
+    ~Plain() { ++g_destroyed; }
+    int value;
+};
+
+// Models an object that owns libuv handles: it can only be recycled once
+// its asynchronous closes have finished.
+struct Closable {
+    static inline ss::CloseCompletion pending;
+    static inline int begun = 0;
+
+    void begin_close(ss::CloseCompletion done) noexcept
+    {
+        ++begun;
+        pending = done;
+    }
+    ~Closable() { ++g_destroyed; }
+};
+
+struct ClosesSynchronously {
+    void begin_close(ss::CloseCompletion done) noexcept { done(); }
+};
+
+void test_acquire_release()
+{
+    ss::ObjectPool<Plain> pool;
+    SS_CHECK(pool.reserve(2));
+    g_destroyed = 0;
+    {
+        auto a = pool.acquire(1);
+        auto b = pool.acquire(2);
+        SS_CHECK(a && b);
+        SS_CHECK((*a)->value == 1 && (*b)->value == 2);
+        SS_CHECK(pool.live() == 2 && pool.full());
+
+        auto c = pool.acquire(3);
+        SS_CHECK(!c && c.error().code == ss::Errc::exhausted);
+    }
+    SS_CHECK(g_destroyed == 2);
+    SS_CHECK(pool.live() == 0 && pool.available() == 2);
+}
+
+void test_reuse_and_handles()
+{
+    ss::ObjectPool<Plain> pool;
+    SS_CHECK(pool.reserve(1));
+    auto first = pool.acquire(10);
+    SS_CHECK(first);
+    ss::PoolPtr<Plain> owner = std::move(first).value();
+    ss::Handle<Plain> h      = pool.handle_of(owner.get());
+    SS_CHECK(h);
+    SS_CHECK(pool.resolve(h) == owner.get());
+
+    owner.reset();
+    SS_CHECK(pool.resolve(h) == nullptr);
+
+    // The slot is reused, but the stale handle must not resolve to the new
+    // occupant.
+    auto second = pool.acquire(20);
+    SS_CHECK(second);
+    SS_CHECK(pool.resolve(h) == nullptr);
+    SS_CHECK(pool.resolve(pool.handle_of(second->get())) == second->get());
+    SS_CHECK(!pool.resolve(ss::Handle<Plain>{}));
+}
+
+void test_async_close()
+{
+    ss::ObjectPool<Closable> pool;
+    SS_CHECK(pool.reserve(1));
+    g_destroyed      = 0;
+    Closable::begun  = 0;
+    auto obj         = pool.acquire();
+    SS_CHECK(obj);
+    ss::Handle<Closable> h = pool.handle_of(obj->get());
+
+    obj->reset();
+    // Ownership is gone, but the memory is not reusable yet.
+    SS_CHECK(Closable::begun == 1);
+    SS_CHECK(g_destroyed == 0);
+    SS_CHECK(pool.closing() == 1 && pool.available() == 0);
+    SS_CHECK(pool.resolve(h) == nullptr);
+    SS_CHECK(!pool.acquire());
+
+    Closable::pending();
+    SS_CHECK(g_destroyed == 1);
+    SS_CHECK(pool.closing() == 0 && pool.available() == 1);
+    {
+        auto again = pool.acquire();
+        SS_CHECK(again);
+    }
+    // The pool refuses to die with a close still in flight, so finish it.
+    SS_CHECK(pool.closing() == 1);
+    Closable::pending();
+    SS_CHECK(pool.closing() == 0);
+}
+
+void test_sync_close()
+{
+    ss::ObjectPool<ClosesSynchronously> pool;
+    SS_CHECK(pool.reserve(1));
+    {
+        auto obj = pool.acquire();
+        SS_CHECK(obj);
+    }
+    SS_CHECK(pool.available() == 1);
+}
+
+void test_reserve_rules()
+{
+    ss::ObjectPool<Plain> pool;
+    SS_CHECK(!pool.acquire(1));
+    SS_CHECK(!pool.reserve(0));
+    SS_CHECK(pool.reserve(4));
+    SS_CHECK(!pool.reserve(4));
+    SS_CHECK(pool.capacity() == 4);
+}
+
+} // namespace
+
+int main()
+{
+    SS_RUN(test_acquire_release);
+    SS_RUN(test_reuse_and_handles);
+    SS_RUN(test_async_close);
+    SS_RUN(test_sync_close);
+    SS_RUN(test_reserve_rules);
+    return 0;
+}
@@ -0,0 +1,79 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#include <memory>
+#include <utility>
+
+#include "core/result.hpp"
+#include "test_helpers.hpp"
+
+namespace {
+
+ss::Result<int> parse_positive(int v)
+{
+    if (v <= 0) {
+        return ss::fail(ss::Errc::invalid);
+    }
+    return v;
+}
+
+ss::Status check_twice(int v)
+{
+    SS_TRY(parse_positive(v));
+    SS_TRY(parse_positive(v - 1));
+    return {};
+}
+
+void test_value_and_error()
+{
+    auto ok = parse_positive(5);
+    SS_CHECK(ok.has_value());
+    SS_CHECK(*ok == 5);
+    SS_CHECK(ok.value_or(0) == 5);
+
+    auto bad = parse_positive(-1);
+    SS_CHECK(!bad);
+    SS_CHECK(bad.error().code == ss::Errc::invalid);
+    SS_CHECK(bad.value_or(42) == 42);
+}
+
+void test_status_and_try()
+{
+    SS_CHECK(check_twice(3).has_value());
+    auto r = check_twice(1);
+    SS_CHECK(!r);
+    SS_CHECK(r.error() == ss::Error{ ss::Errc::invalid, 0 });
+}
+
+void test_move_only_value()
+{
+    auto make = []() -> ss::Result<std::unique_ptr<int> > { return std::make_unique<int>(7); };
+    auto r    = make();
+    SS_CHECK(r);
+    std::unique_ptr<int> p = std::move(r).value();
+    SS_CHECK(*p == 7);
+}
+
+void test_monadic()
+{
+    auto doubled = parse_positive(4).transform([](int v) { return v * 2; });
+    SS_CHECK(doubled && *doubled == 8);
+    auto chained = parse_positive(4).and_then([](int v) { return parse_positive(v - 10); });
+    SS_CHECK(!chained && chained.error().code == ss::Errc::invalid);
+}
+
+void test_sys_errno()
+{
+    ss::Result<int> r = ss::fail(ss::Errc::io, 104);
+    SS_CHECK(r.error().sys == 104);
+}
+
+} // namespace
+
+int main()
+{
+    SS_RUN(test_value_and_error);
+    SS_RUN(test_status_and_try);
+    SS_RUN(test_move_only_value);
+    SS_RUN(test_monadic);
+    SS_RUN(test_sys_errno);
+    return 0;
+}
@@ -0,0 +1,26 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_TEST_HELPERS_HPP
+#define SS_TEST_HELPERS_HPP
+
+// Minimal harness for the C++ unit tests. Tests are built without
+// exceptions, so a failed check reports its location and aborts.
+
+#include <cstdio>
+#include <cstdlib>
+
+#define SS_CHECK(...)                                                         \
+    do {                                                                      \
+        if (!(__VA_ARGS__)) {                                                      \
+            std::fprintf(stderr, "%s:%d: check failed: %s\n", __FILE__,       \
+                         __LINE__, #__VA_ARGS__);                                 \
+            std::abort();                                                     \
+        }                                                                     \
+    } while (0)
+
+#define SS_RUN(test)                                                          \
+    do {                                                                      \
+        test();                                                               \
+        std::printf("ok %s\n", #test);                                        \
+    } while (0)
+
+#endif