yams/tools/fuzzing/fuzz_ipc_roundtrip.cpp
Trevon ae15bf317a
perf(ingest): harden daemon batching and shutdown (#58)
* fix(daemon): harden model state ownership

Keep ModelPoolConfig single-defined and serialize telemetry snapshots to avoid ODR drift and torn concurrent reads.

* refactor(daemon): consolidate ingest pollers

Require the wake-timer contract so every post-ingest stage shares one event-driven poller configuration.

* refactor(daemon): unify repair execution

Share admission, operation ordering, cancellation, progress, and result aggregation between synchronous and asynchronous repair paths.

* refactor(daemon): unify runtime tuning

Resolve startup and reload settings through one typed path so parsing, precedence, and range validation cannot drift.

* refactor(daemon)!: remove dead IPC surfaces

Delete unreferenced framing, cancellation, and compatibility paths while keeping the active streaming protocol intact.

BREAKING CHANGE: callers using connectAsync or legacy MessageFramer and FrameReader helpers must migrate to connect, frame_message_into, feed, and get_frame.

* refactor(daemon): hide test hooks

Preserve test-library linkage without exposing daemon-only probes through production headers or external symbols.

* fix(daemon): avoid replaying successful adds

A fresh-connection transport retry reuses the response handler. Clear its stale EOF error when the retry returns a valid add response so the outer retry loop does not enqueue the document again.

* bench(ingest): enforce lossless pipeline runs

Separate admission, storage, drain, enrichment, and searchability timing. Decision-grade plans now reject drops, duplicate processing, incomplete pipelines, and experiment identity drift.

* bench(search): expose cold seed ingest timings

Pair generalized-memory quality arms with admission, storage, drain, searchability, and enrichment timing from their shared cold seed. Keep the observation distinct from repeated throughput measurements.

* perf(ingest): batch deferred content indexing

Directory ingestion already schedules content work through post-ingest. Avoid duplicate inline indexing and coalesce normal-priority trickles within a bounded window.

* feat(metadata): expose insert writer metrics

Separate queue, batch-apply, and transaction time so ingestion runs can attribute connection pressure without changing write semantics.

* fix(daemon): retry transient refused connects

The daemon socket can be bound before its listener is ready during startup. Reuse the existing bounded retry budget for that refusal window.

* bench(ingest): attribute submission path costs

Decision-grade runs need stable coalescing identity and separate queue, apply, transaction, and estimated connection-wait costs.

* perf(ingest): batch durable document storage

Commit chunks, manifests, references, and metadata in bounded batches while preserving publication only after durable reference success.

* perf(ingest): coalesce bulk post-processing

Use a 16-item bulk cap while keeping latency-sensitive daemon ingestion on the existing four-item priority quota.

* bench(ingest): record post-ingest batch size

Include the effective cap in metrics and experiment identity so runs with different batching policies cannot be compared as one system.

* perf(ingest): batch queued storage waves

Preserve single-document fallback semantics and execute inline when the coordinator has one worker to avoid self-deadlock.

* perf(ingest): route bulk enrichment batches

Keep quota-sized interactive groups on the RPC lane while sending larger lossless groups through bounded coalescing.

* test(daemon): initialize task fixtures

* perf(ingest): batch post-ingest dispatch

* perf(ingest): widen directory storage waves

Borrow from a full wave when one document would otherwise fall through to single-document storage.

* perf(storage): bound batch writer fanout

Reuse a fixed worker set so batch size no longer creates one thread and payload copy per object.

* perf(ingest): align storage batch drains

Dispatch one configured batch through storage so intermediate waves do not fragment downstream metadata work.

* fix(ingest): commit metadata state atomically

Fold first-version, extracted-title, and embedding-completion state
into existing transactions.

Read committed extraction state so duplicate-content tasks cannot
double-count corpus counters.

* bench(ingest): measure production submission

Drive pipelined files through DocumentIngestionService.
Export dispatch, fingerprint, and enqueue costs through xplan.

* fix(xplan): preserve native failures

Report child signals and bounded stderr before readiness checks so
assertions are not masked as missing traces.

* fix(daemon): preserve embedding success hashes

* fix(daemon): serialize pooled socket shutdown

* fix(daemon): harden batch publication shutdown

* fix(storage): use compat jthread shim

* fix(daemon): avoid poller template shadowing

* fix(search): keep vector pipeline test enabled

* fix: harden shutdown and fallback paths

Demote expected shutdown SQL interruptions, ignore readonly FTS integrity noise, avoid stopped-executor socket hangs, and preserve metadata writer fallback outcomes.
2026-07-17 07:04:30 -04:00

129 lines
4.1 KiB
C++

// SPDX-License-Identifier: GPL-3.0-or-later
// IPC Roundtrip Fuzzer - Targets CLI<->daemon framing and payload encode/decode
#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <span>
#include <vector>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_future.hpp>
#include <yams/daemon/ipc/message_framing.h>
#include <yams/daemon/ipc/proto_serializer.h>
#include <yams/daemon/ipc/request_handler.h>
using namespace yams::daemon;
namespace {
class FuzzRoundtripProcessor : public RequestProcessor {
public:
boost::asio::awaitable<Response> process(const Request& request) override {
if (std::holds_alternative<PingRequest>(request)) {
PongResponse pong;
pong.serverTime = std::chrono::steady_clock::now();
pong.roundTrip = std::chrono::milliseconds{0};
co_return Response{std::in_place_type<PongResponse>, std::move(pong)};
}
if (std::holds_alternative<StatusRequest>(request)) {
StatusResponse status{};
status.running = true;
status.ready = true;
status.version = "fuzz";
co_return Response{std::in_place_type<StatusResponse>, std::move(status)};
}
co_return SuccessResponse{"fuzz: ok"};
}
};
std::vector<uint8_t> run_handler_roundtrip(std::span<const uint8_t> payload,
const RequestHandler::Config& config) {
boost::asio::io_context io;
auto processor = std::make_shared<FuzzRoundtripProcessor>();
RequestHandler handler(processor, config);
std::vector<uint8_t> request_data(payload.begin(), payload.end());
auto fut = boost::asio::co_spawn(
io, handler.handle_request(std::move(request_data), yams::compat::stop_token{}),
boost::asio::use_future);
io.run();
return fut.get();
}
} // namespace
extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
if (size < MessageFramer::HEADER_SIZE) {
return 0;
}
FrameReader reader;
MessageFramer framer;
const size_t chunk = std::max<size_t>(1, size / 4);
size_t offset = 0;
while (offset < size) {
const size_t remaining = size - offset;
const size_t to_feed = std::min(remaining, chunk);
const auto feed_result = reader.feed(data + offset, to_feed);
offset += feed_result.consumed;
if (reader.has_frame()) {
auto frame_result = reader.get_frame();
if (!frame_result)
continue;
const auto& frame = frame_result.value();
if (frame.size() < MessageFramer::HEADER_SIZE) {
continue;
}
auto header_result = framer.parse_header(frame);
if (!header_result) {
continue;
}
const auto payload_size = header_result.value().payload_size;
if (payload_size == 0 || frame.size() < MessageFramer::HEADER_SIZE + payload_size) {
continue;
}
const auto payload =
std::span<const uint8_t>(frame.data() + MessageFramer::HEADER_SIZE, payload_size);
RequestHandler::Config config;
config.enable_streaming = (data[0] & 1) != 0;
config.enable_multiplexing = (data[0] & 2) != 0;
config.max_frame_size = 1 * 1024 * 1024;
std::vector<uint8_t> response_payload;
try {
response_payload = run_handler_roundtrip(payload, config);
} catch (...) {
continue;
}
auto response_msg = ProtoSerializer::decode_payload(response_payload);
if (!response_msg) {
continue;
}
auto response_frame = framer.frame_message(response_msg.value());
if (!response_frame) {
continue;
}
auto parsed_response = framer.parse_frame(response_frame.value());
(void)parsed_response;
}
if (feed_result.consumed == 0) {
break;
}
}
return 0;
}