Release of fifo and mutexQueue to OSS with bio refactored - #3

…lkey-io#3174)

I was working on ASAN large memory tests when I countered this issue.

The issue was that the hardcoded `999` key could land in an early
bucket. Then shrink rehash could finish early, and later inserts could
trigger a new expansion rehash, resetting rehash_idx low. The test now
picks the survivor key dynamically as the key mapped to the highest
bucket index.

```
[test_hashtable.c] Memory leak detected of 336 bytes
=================================================================
==3901==ERROR: LeakSanitizer: detected memory leaks

Direct leak of 80 byte(s) in 1 object(s) allocated from:
    #0 0x7fb0556fd9c7 in malloc ../../../../src/libsanitizer/asan/asan_malloc_linux.cpp:69
    #1 0x563bfdf4c47d in ztrymalloc_usable_internal /home/runner/work/valkey/valkey/src/zmalloc.c:156
    #2 0x563bfdf4c47d in valkey_malloc /home/runner/work/valkey/valkey/src/zmalloc.c:185
    #3 0x563bfdd42eaf in hashtableCreate /home/runner/work/valkey/valkey/src/hashtable.c:1217
    #4 0x563bfdaa1cbf in test_empty_buckets_rehashing unit/test_hashtable.c:232
    #5 0x563bfdae772b in runTestSuite unit/test_main.c:36
    #6 0x563bfda86b20 in main unit/test_main.c:108
    #7 0x7fb05522a1c9  (/lib/x86_64-linux-gnu/libc.so.6+0x2a1c9) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #8 0x7fb05522a28a in __libc_start_main (/lib/x86_64-linux-gnu/libc.so.6+0x2a28a) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #9 0x563bfda8a5c4 in _start (/home/runner/work/valkey/valkey/src/valkey-unit-tests+0x17c5c4) (BuildId: 44cfc183e6e82e499bcc9f6adc094d7f774ee9d2)

Indirect leak of 128 byte(s) in 1 object(s) allocated from:
    #0 0x7fb0556fd340 in calloc ../../../../src/libsanitizer/asan/asan_malloc_linux.cpp:77
    #1 0x563bfdf4c922 in ztrycalloc_usable_internal /home/runner/work/valkey/valkey/src/zmalloc.c:214
    #2 0x563bfdf4c922 in valkey_calloc /home/runner/work/valkey/valkey/src/zmalloc.c:257
    #3 0x563bfdd40967 in resize /home/runner/work/valkey/valkey/src/hashtable.c:741
    #4 0x563bfdd45eb1 in hashtableExpandIfNeeded /home/runner/work/valkey/valkey/src/hashtable.c:1446
    #5 0x563bfdd45eb1 in hashtableExpandIfNeeded /home/runner/work/valkey/valkey/src/hashtable.c:1433
    #6 0x563bfdd45eb1 in insert /home/runner/work/valkey/valkey/src/hashtable.c:1041
    #7 0x563bfdd45eb1 in hashtableAddOrFind /home/runner/work/valkey/valkey/src/hashtable.c:1554
    #8 0x563bfdd45eb1 in hashtableAdd /home/runner/work/valkey/valkey/src/hashtable.c:1539
    #9 0x563bfdaa1e3b in test_empty_buckets_rehashing unit/test_hashtable.c:254
    #10 0x563bfdae772b in runTestSuite unit/test_main.c:36
    #11 0x563bfda86b20 in main unit/test_main.c:108
    #12 0x7fb05522a1c9  (/lib/x86_64-linux-gnu/libc.so.6+0x2a1c9) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #13 0x7fb05522a28a in __libc_start_main (/lib/x86_64-linux-gnu/libc.so.6+0x2a28a) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #14 0x563bfda8a5c4 in _start (/home/runner/work/valkey/valkey/src/valkey-unit-tests+0x17c5c4) (BuildId: 44cfc183e6e82e499bcc9f6adc094d7f774ee9d2)

Indirect leak of 64 byte(s) in 1 object(s) allocated from:
    #0 0x7fb0556fd340 in calloc ../../../../src/libsanitizer/asan/asan_malloc_linux.cpp:77
    #1 0x563bfdf4c922 in ztrycalloc_usable_internal /home/runner/work/valkey/valkey/src/zmalloc.c:214
    #2 0x563bfdf4c922 in valkey_calloc /home/runner/work/valkey/valkey/src/zmalloc.c:257
    #3 0x563bfdd3f553 in bucketConvertToChained /home/runner/work/valkey/valkey/src/hashtable.c:908
    #4 0x563bfdd3f553 in findBucketForInsert /home/runner/work/valkey/valkey/src/hashtable.c:1021
    #5 0x563bfdd45d9e in insert /home/runner/work/valkey/valkey/src/hashtable.c:1045
    #6 0x563bfdd45d9e in hashtableAddOrFind /home/runner/work/valkey/valkey/src/hashtable.c:1554
    #7 0x563bfdd45d9e in hashtableAdd /home/runner/work/valkey/valkey/src/hashtable.c:1539
    #8 0x563bfdaa1e3b in test_empty_buckets_rehashing unit/test_hashtable.c:254
    #9 0x563bfdae772b in runTestSuite unit/test_main.c:36
    #10 0x563bfda86b20 in main unit/test_main.c:108
    #11 0x7fb05522a1c9  (/lib/x86_64-linux-gnu/libc.so.6+0x2a1c9) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #12 0x7fb05522a28a in __libc_start_main (/lib/x86_64-linux-gnu/libc.so.6+0x2a28a) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #13 0x563bfda8a5c4 in _start (/home/runner/work/valkey/valkey/src/valkey-unit-tests+0x17c5c4) (BuildId: 44cfc183e6e82e499bcc9f6adc094d7f774ee9d2)

Indirect leak of 64 byte(s) in 1 object(s) allocated from:
    #0 0x7fb0556fd340 in calloc ../../../../src/libsanitizer/asan/asan_malloc_linux.cpp:77
    #1 0x563bfdf4c922 in ztrycalloc_usable_internal /home/runner/work/valkey/valkey/src/zmalloc.c:214
    #2 0x563bfdf4c922 in valkey_calloc /home/runner/work/valkey/valkey/src/zmalloc.c:257
    #3 0x563bfdd40967 in resize /home/runner/work/valkey/valkey/src/hashtable.c:741
    #4 0x563bfdaa1df8 in test_empty_buckets_rehashing unit/test_hashtable.c:248
    #5 0x563bfdae772b in runTestSuite unit/test_main.c:36
    #6 0x563bfda86b20 in main unit/test_main.c:108
    #7 0x7fb05522a1c9  (/lib/x86_64-linux-gnu/libc.so.6+0x2a1c9) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #8 0x7fb05522a28a in __libc_start_main (/lib/x86_64-linux-gnu/libc.so.6+0x2a28a) (BuildId: 274eec488d230825a136fa9c4d85370fed7a0a5e)
    #9 0x563bfda8a5c4 in _start (/home/runner/work/valkey/valkey/src/valkey-unit-tests+0x17c5c4) (BuildId: 44cfc183e6e82e499bcc9f6adc094d7f774ee9d2)

SUMMARY: AddressSanitizer: 336 byte(s) leaked in 4 allocation(s).
```

Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>

CI caught ip and name SDS allocations being leaked in
fetchClusterConfiguration. The ip SDS was copied again via sdsnew()
before being passed to createClusterNode(), leaking the original. The
name SDS was leaked when the node already existed in the dict.

Free ip and name on all exit paths in fetchClusterConfiguration. Remove
stale guard in freeClusterNode, no longer needed since valkey-io#1392

CI Error -
```
Direct leak of 33 byte(s) in 3 object(s) allocated from:
      #0 0x7f4c3a0fd9c7 in malloc ../../../../src/libsanitizer/asan/asan_malloc_linux.cpp:69
      #1 0x5564620c124a in ztrymalloc_usable_internal /home/runner/work/valkey/valkey/src/zmalloc.c:172
      #2 0x5564620c124a in zmalloc_usable /home/runner/work/valkey/valkey/src/zmalloc.c:268
      #3 0x5564620dfbe6 in _sdsnewlen.constprop.0 /home/runner/work/valkey/valkey/src/sds.c:102
      #4 0x556462050996 in sdsnewlen /home/runner/work/valkey/valkey/src/sds.c:169
      #5 0x556462050996 in sdsnew /home/runner/work/valkey/valkey/src/sds.c:185
      #6 0x556462050996 in fetchClusterConfiguration /home/runner/work/valkey/valkey/src/valkey-benchmark.c:1477
```

Issue was reproduceable locally using `leaks --atExit`

Signed-off-by: nmvk <r@nmvk.com>

)

### Problem

The bug was originally introduced by valkey-io#1737

In tls.c, we normally avoid consolidating IO vectors into a buffer if
they exceed a capped amount (`NET_MAX_WRITES_PER_EVENT`, which is 64KB).
However, on OpenSSL write errors, we must never trigger a subsequent
write that is smaller than the previous failed write. In this path, we
must consolidate the IO vectors to ensure the write is at least as large
as the last write we made on the socket:

```c
    char buf[iov_bytes_len];
    size_t offset = 0;
    for (int i = 0; i < iovcnt && offset < iov_bytes_len; i++) {
        memcpy(buf + offset, iov[i].iov_base, iov[i].iov_len);
```

However, `iov_bytes_len` can easily exceed the stack boundaries if one
of the IO vectors is large (e.g. a large key read), leading to a crash:

```
Thread 1 (Thread 0x7ffff7b65700 (LWP 191) "valkey-server"):
#0  0x00005555557a04bc in connTLSWritev (conn_=0x7ffff6e518c0, iov=<optimized out>, iovcnt=5) at src/tls.c:1713
#1  0x00005555556d2a21 in connWritev (conn=<optimized out>, iov=0x7fffffff9240, iovcnt=5) at src/connection.h:256
#2  writevToClient (c=<optimized out>) at src/networking.c:2814
#3  _writeToClient (c=<optimized out>) at src/networking.c:2868
#4  0x00005555556d326c in writeToClient (c=0x7ffff6e7a780) at src/networking.c:3117
#5  writeToClient (c=0x7ffff6e7a780) at src/networking.c:3108
#6  sendReplyToClient (conn=<optimized out>) at src/networking.c:3127
#7  0x00005555557a0ffd in callHandler (conn=0x7ffff6e518c0, handler=<optimized out>) at src/connhelpers.h:79
#8  tlsHandleEvent (conn=0x7ffff6e518c0, mask=<optimized out>) at src/tls.c:1507
#9  0x00005555555fbad5 in aeProcessEvents (flags=27, eventLoop=0x7ffff6e45f80) at src/ae.c:504
#10 aeMain (eventLoop=0x7ffff6e45f80) at src/ae.c:543
#11 0x00005555555eb83a in main (argc=2, argv=0x7fffffffd588) at src/server.c:7833
```

### Fix

To fix this, we simply avoid allocating large buffers on the stack by
using a fixed-size stack buffer `char buf[NET_MAX_WRITES_PER_EVENT]`
(64KB) and performing partial copies:
1. Calculate total length of all `iov` blocks to decide if we can use
one-by-one writing.
2. If total length is larger than `NET_MAX_WRITES_PER_EVENT` (64KB) and
`iov[0]` is large enough to satisfy `last_failed_write_data_len`, write
blocks sequentially one-by-one.
3. Otherwise (combine path), copy data from `iov` into a fixed-size
stack buffer of `NET_MAX_WRITES_PER_EVENT` (64KB), capping the copy at
64KB (partial copy if it exceeds).
4. Verify that we copied at least `last_failed_write_data_len` to
satisfy OpenSSL retry constraint.
5. Write the combined buffer and return the number of bytes copied
(Valkey connection layer will handle the remaining data as a partial
write retry).

Important: this should keep the existing "write must be larger than last
failed write" invariant, considering: 1) if the last write went to the
consolidated buffer path, it would be at most `NET_MAX_WRITES_PER_EVENT`
2) if the last write did not go through the consolidated buffer path, it
will be at the front of the `iov`, and therefore our `iov[0] >=
last_failed_write_data_len` check will always pass (it is the same
write).

Finally, since we added IO threaded TLS writes, the stack limit may be
even smaller for the thread (depending on the system). We work around
this in bio threads, and I am copying that workaround to IO threads to
ensure the stack is not smaller than `NET_MAX_WRITES_PER_EVENT`.

### Testing

A regression test was appended to the existing TLS test suite in
tests/unit/tls.tcl. We need to add a new debug subcommand `DEBUG
FORCE-TLS-WRITE-ERROR <0 or 1>` to deterministically trigger OpenSSL
write failures. With this flow, we write a command with a small response
and force it to trigger an IO error, then we write a command with a
larger response and resolve the IO errors. The response to the large
command will accumulate in the buffer, and we will attempt to write out
the full output buffer for the client, which now includes a massive
final IO vector. In the old code, this would trigger a stack overflow as
we try to allocate O(megabytes) on the stack.

---------

Signed-off-by: Jacob Murphy <jkmurphy@google.com>
Co-authored-by: Ran Shidlansik <ranshid@amazon.com>

Fixes valkey-io#3812

Fixes a segfault when running a Lua module build under alpine/musl. The
dynamic linker was resolving the `getMonotonicUs` function in
`libvalkeylua` to `getMonotonicUs` in the main binary, which is a
function pointer. Calls to `getMonoticUs` via the PLT were resolving to
the variable in `.bss`, causing a segfault.
I marked the functions as static to fix this, and I imagine they should
have been static in the first place (correct me if I'm wrong).

Stack trace:
```
[ #3] 0x7f968ed83de2 <__restore_rt>
[ #4] 0x561cae23dd30 <getMonotonicUs>
[*#5] 0x7f968ebcd186 <luaFunctionLibraryCreate+0xd6>
```

```
gdb> vmmap 0x561cae23dd30
Start              End                Size               Offset             Perm Path
0x0000561cae1d8000 0x0000561cae25d000 0x0000000000085000 0x00000000002fe000 rw- ./src/valkey-server +0x65d30  <-  $rcx
```

This issue only occurs on alpine because the musl dynamic linker doesn't
support the [`RTLD_DEEPBIND`
option](https://github.com/valkey-io/valkey/blob/unstable/src/module.c#L13483).
The glibc dynamic linker was resolving the `getMonotonicUs` in
`libvalkeylua` and not the "global" symbol which is a function pointer.

---
## Reproduce

This is script to spawn a docker alpine container to reproduce the
crash, based on the instructions in valkey-io#3812
```bash
#!/usr/bin/env bash
cleanup() {
    echo "deleting container"
    docker rm -f valkey-repro
}

trap cleanup EXIT

echo "create alpine container"

rm -f dump.rdb

docker run -d \
	--name valkey-repro \
	--rm \
	-v "$PWD":"$PWD" \
	-w "$PWD" \
	-it \
	alpine:3.23.5 \
    sleep infinity


echo "install deps"
docker exec valkey-repro sh -c 'apk add build-base linux-headers openssl-dev tcl'

echo "build"
docker exec valkey-repro sh -c 'make distclean && make clean'
docker exec valkey-repro sh -c 'make BUILD_LUA=module USE_JEMALLOC=no MALLOC=libc -j25'

echo "waiting for server to send command"
(
    until docker exec valkey-repro test -S /tmp/valkey.sock; do
        sleep 0.05
    done

    echo "======== server is up, sending command ========="

    docker exec valkey-repro \
        ./src/valkey-cli \
        -s /tmp/valkey.sock \
        FUNCTION LOAD "#!LUA name=test2
        server.register_function('test2', function(KEYS, ARGV)
            return 'hello'
        end)"
) &

echo "start server"

docker exec valkey-repro \
    ./src/valkey-server \
    --port 0 \
    --unixsocket /tmp/valkey.sock \
    --loadmodule ./src/modules/lua/libvalkeylua.so

echo "opening shell in container"
docker exec -it valkey-repro sh

```

Signed-off-by: mohammedgqudah <mohghq@gmail.com>

During a disk-based full sync that transcodes the RDB (wire codec
differs from the replica's `rdbcompression`), any bytes after the end of
the primary's VCS frame are left pending in the push reader, and the
next transcoder write trips `serverAssert(len == 0 || !resuming)` and
crashes the replica. A well-formed payload never has such bytes, so this
is reachable only with a malformed or corrupted payload, but it should
fail the sync, not the process. This fails the transcode when the frame
ends with input left over, so the replica aborts the sync and retries.

<details>
<summary>AI generated details</summary>

## Problem

`rdbTranscoderRun()` breaks out of its feed loop on
`STREAM_PUSH_READER_FRAME_DONE` (`src/rdb_transcoder.c:141`), on the
assumption stated in its comment that the caller already excluded
trailing bytes. `streamPushReaderFeed()` stashes whatever follows the
frame end in `pending_input` (`src/compression_stream.c:616-618`). After
the frame ends the decompressor consumes nothing
(`src/compression.c:161`), so later input is stashed too. The next feed
with input then asserts at `src/compression_stream.c:593`.

LZ4 source, plaintext target:

| Bytes after the frame | Before | After |
|---|---|---|
| none | OK | OK |
| in the same write as the frame end | silently dropped, transcode
succeeds | transcode fails |
| in one later write | stashed, transcode succeeds | transcode fails |
| in two or more later writes | `serverAssert` crash | transcode fails |

`replicaReceiveRDBFromPrimaryToDisk()` calls `rdbTranscoderWrite()` once
per read for size-framed transfers and up to twice per read for
`$EOF`-marked ones (`src/replication.c:3199-3210`), so trailing bytes
that span a read boundary reach the crash. The verbatim route (source
codec equals target) never decodes; it still writes trailing bytes to
disk, where the loader stops at the frame end and ignores them.

Introduced in valkey-io#4749 (`7fe7ca45f` on unstable, `4216cddd7` on 9.2). Not
in any release tag.

## Can a well-behaved primary send trailing bytes?

No. The receiver holds back the trailing `RDB_EOF_MARK_SIZE` bytes, so
the `$EOF` mark is never fed (`src/replication.c:3193-3214`). The
primary writes exactly one codec frame per RDB (`rdbSaveRio()`,
`src/rdb.c:1675`), and the diskless `$EOF` trailer is written after it
returns and the writer is detached (`src/rdb.c:1631`).

## Alternative considered

Ignore the trailing bytes instead of failing. The on-disk loader stops
at the frame end
(`CompressionTest.streamReaderStopsBeforeTrailingBytes`), so the
verbatim route already tolerates them. That loses: nothing legitimate
produces them, the transcoder's own contract says there are none, and it
needs a frame-done flag plus discarding pending input on every later
write rather than one check. Failing the sync costs a retry on a payload
that is already wrong.

## Behavior change

Trailing bytes that arrive with the frame end used to be dropped and the
sync succeeded. They now fail the sync. Only malformed payloads are
affected.

## Testing

`CompressionTest.rdbTranscoderRejectsBytesAfterSourceFrame` in
`src/unit/test_compression.cpp` drives the transcoder directly for every
source/target pair that converts, with the trailer in the frame's write
and split across two later writes. There were no unit tests for the
transcoder before. With the `src/rdb_transcoder.c` hunk reverted:

```
[ RUN      ] CompressionTest.rdbTranscoderRejectsBytesAfterSourceFrame
test_compression.cpp:1830: Failure
Expected equality of these values:
  rdbTranscoderWrite(&t, with_trailer, sdslen(with_trailer))
    Which is: 0
  -1
Google Test trace:
test_compression.cpp:1824: src=2 target=0 same_write=true
...
test_compression.cpp:1835: Failure
Expected equality of these values:
  rdbTranscoderWrite(&t, trailer, 3)
    Which is: 0
  -1
Google Test trace:
test_compression.cpp:1824: src=2 target=0 same_write=false
```

The second trailer write then dies in the assert:

```
frame #3: _serverAssert(estr="len == 0 || !resuming", file="compression_stream.c", line=593) at debug.c:1180:22
frame #4: streamPushReaderFeed(..., len=4, ...) at compression_stream.c:593:5
frame #5: rdbTranscoderRun + 180
frame #6: rdbTranscoderWrite + 392
frame #7: CompressionTest_rdbTranscoderRejectsBytesAfterSourceFrame_Test::TestBody() at test_compression.cpp:1836:21
```

</details>

This was generated by AI but verified, with love, by a human.

---------

Signed-off-by: Madelyn Olson <madelyneolson@gmail.com>
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
Co-authored-by: Sarthak Aggarwal <sarthagg@amazon.com>