cproof/tests/bench/bench_long_messages.c

// SPDX-License-Identifier: GPL-3.0-or-later
//
// This file is part of CProof.
// See LICENSE for the full GPLv3 text and the special OpenSSL linking exception.

/*
 * bench_long_messages.c
 * vim: expandtab:ts=4:sts=4:sw=4
 *
 * L1–L14: long-message stress tests for the flat-file backend.
 *
 *   L1   1 KB body                   roundtrip identity
 *   L2   10 KB body                  roundtrip identity
 *   L3   100 KB body                 roundtrip identity
 *   L4   1 MB body                   roundtrip identity + write/read time
 *   L5   5 MB body                   roundtrip identity
 *   L6   9.9 MB body                 just under FF_MAX_LINE_LEN (10 MB)
 *   L7   10 MB + 1 body              expect ff_readline reject (returns "")
 *   L8   100 KB with embedded \n×1k  escape stress (each \n -> \\n doubles)
 *   L9   100 KB with `|` × 50k       not in metadata, parser shouldn't choke
 *   L10  100 KB emoji-only           4-byte UTF-8 codepoints
 *   L11  100 × 1 KB roundtrip        sanity baseline
 *   L12  pagination — last 100 with 5×1MB bodies
 *   L13  verify-pass on 100×1MB
 *   L14  rapid append: 1000×100KB
 *
 * Each test:
 *   1. Generates N messages with a specific body_size and content_pattern.
 *   2. Writes them via ff_write_line into a fresh temp file.
 *   3. Reads them back via ff_readline + ff_parse_line.
 *   4. Asserts body length / content invariants.
 *   5. Writes a CSV row: L#, body_size, n, write_ms, read_ms, peak_rss
 */

#include "config.h"

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>

#include <glib.h>

#include "bench_common.h"
#include "bench_csv.h"
#include "database_flatfile.h"

// ---------------------------------------------------------------------------
// CLI

typedef struct
{
    const char* tmp_dir;
    const char* csv_path;
    const char* tests; // comma list or "all"
} cli_t;

static int
parse_args(int argc, char** argv, cli_t* c)
{
    c->tmp_dir = "/tmp/cproof-bench-longmsg";
    c->csv_path = NULL;
    c->tests = "all";
    for (int i = 1; i < argc; i++) {
        const char* a = argv[i];
        if (strncmp(a, "--tmp=", 6) == 0) c->tmp_dir = a + 6;
        else if (strncmp(a, "--csv=", 6) == 0) c->csv_path = a + 6;
        else if (strncmp(a, "--tests=", 8) == 0) c->tests = a + 8;
        else {
            fprintf(stderr, "unknown option: %s\n", a);
            return -1;
        }
    }
    return 0;
}

static int
test_enabled(const char* list, const char* name)
{
    if (!list || g_strcmp0(list, "all") == 0)
        return 1;
    auto_gcharv gchar** parts = g_strsplit(list, ",", -1);
    for (int i = 0; parts[i]; i++) {
        if (g_strcmp0(g_strstrip(parts[i]), name) == 0)
            return 1;
    }
    return 0;
}

// ---------------------------------------------------------------------------
// Body factories

typedef enum
{
    PAT_FILLER,        // deterministic alpha-num
    PAT_EMBEDDED_LF,   // filler with newlines every 100 bytes
    PAT_EMBEDDED_PIPE, // filler with '|' sprinkled
    PAT_EMOJI,         // repeated 4-byte UTF-8 emoji
} pattern_t;

static char*
make_body(size_t target_len, pattern_t pat)
{
    if (pat == PAT_EMOJI) {
        // 4 bytes per "🚀" (U+1F680). Round target down to 4-byte boundary.
        size_t n_emoji = target_len / 4;
        size_t bytes = n_emoji * 4;
        char* buf = g_malloc(bytes + 1);
        const char emoji[5] = { (char)0xF0, (char)0x9F, (char)0x9A, (char)0x80, 0 };
        for (size_t i = 0; i < n_emoji; i++)
            memcpy(buf + i * 4, emoji, 4);
        buf[bytes] = '\0';
        return buf;
    }

    char* buf = g_malloc(target_len + 1);
    static const char alpha[] = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOP";
    static const int alpha_n = sizeof(alpha) - 1;
    for (size_t i = 0; i < target_len; i++)
        buf[i] = alpha[i % alpha_n];

    if (pat == PAT_EMBEDDED_LF) {
        for (size_t i = 100; i < target_len; i += 100)
            buf[i] = '\n';
    } else if (pat == PAT_EMBEDDED_PIPE) {
        // sprinkle ~50k pipes uniformly across 100 KB
        size_t step = target_len > 50000 ? target_len / 50000 : 2;
        for (size_t i = 0; i < target_len; i += step)
            buf[i] = '|';
    }
    buf[target_len] = '\0';
    return buf;
}

// ---------------------------------------------------------------------------
// Roundtrip primitive

typedef struct
{
    char* path;
    int64_t bytes_written;
    double write_ms;
    double read_ms;
    int64_t parsed;
    int64_t mismatched_len;
    int64_t parse_failures;
    long peak_rss_kb;
} roundtrip_result_t;

static void
roundtrip_cleanup(roundtrip_result_t* r)
{
    if (!r) return;
    if (r->path && r->path[0])
        unlink(r->path);
    g_free(r->path);
    memset(r, 0, sizeof(*r));
}

static int
write_n_messages(const char* path, int n, size_t body_len, pattern_t pat,
                 int64_t* bytes_written_out)
{
    FILE* fp = fopen(path, "w");
    if (!fp) return -1;
    fprintf(fp, "%s", FLATFILE_HEADER);

    GDateTime* base = g_date_time_new_utc(2025, 6, 15, 12, 0, 0.0);
    for (int i = 0; i < n; i++) {
        GDateTime* t = g_date_time_add_seconds(base, i);
        auto_gchar gchar* iso = g_date_time_format_iso8601(t);
        g_date_time_unref(t);
        auto_gchar gchar* sid = g_strdup_printf("long-%d-%08x", i,
                                                (unsigned)(rand_r(&(unsigned){i + 1})));
        auto_gchar gchar* body = make_body(body_len, pat);
        ff_write_line(fp, iso, "chat", "none",
                      sid, NULL, NULL,
                      "alice@bench.example", "phone",
                      NULL, NULL, -1,
                      body);
    }
    g_date_time_unref(base);
    fflush(fp);
    if (bytes_written_out) {
        struct stat st;
        if (fstat(fileno(fp), &st) == 0)
            *bytes_written_out = st.st_size;
    }
    fclose(fp);
    return 0;
}

static int
read_and_validate(const char* path, size_t expected_body_len, int n,
                  int64_t* parsed_out, int64_t* mismatched_out, int64_t* parse_fail_out)
{
    FILE* fp = fopen(path, "r");
    if (!fp) return -1;
    ff_skip_bom(fp);

    int64_t parsed = 0, mismatched = 0, failures = 0;
    char* buf;
    while ((buf = ff_readline(fp, NULL)) != NULL) {
        if (buf[0] == '\0' || buf[0] == '#') {
            free(buf);
            continue;
        }
        ff_parsed_line_t* pl = ff_parse_line(buf);
        free(buf);
        if (!pl) {
            failures++;
            continue;
        }
        parsed++;
        if (expected_body_len > 0 && pl->message
            && strlen(pl->message) != expected_body_len) {
            mismatched++;
        }
        ff_parsed_line_free(pl);
    }
    fclose(fp);
    if (parsed_out) *parsed_out = parsed;
    if (mismatched_out) *mismatched_out = mismatched;
    if (parse_fail_out) *parse_fail_out = failures;
    (void)n;
    return 0;
}

static void
run_roundtrip(const char* tmp_dir, const char* tag,
              int n, size_t body_len, pattern_t pat,
              roundtrip_result_t* out)
{
    out->path = g_strdup_printf("%s/%s.log", tmp_dir, tag);

    double t0 = bench_now_ms();
    if (write_n_messages(out->path, n, body_len, pat, &out->bytes_written) != 0) {
        fprintf(stderr, "  %s: WRITE FAILED\n", tag);
        return;
    }
    out->write_ms = bench_now_ms() - t0;

    bench_drop_page_cache(out->path);
    t0 = bench_now_ms();
    read_and_validate(out->path, body_len, n,
                      &out->parsed, &out->mismatched_len, &out->parse_failures);
    out->read_ms = bench_now_ms() - t0;
    out->peak_rss_kb = bench_peak_rss_kb();
}

static void
csv_emit(const char* csv_path, const char* tag, size_t body_len, int n,
         const roundtrip_result_t* r, const char* note)
{
    if (!csv_path) return;
    auto_gchar gchar* full_note = g_strdup_printf(
        "n=%d body=%zu write=%.1fms read=%.1fms parsed=%" PRId64
        " mismatch=%" PRId64 " parse_fail=%" PRId64 " %s",
        n, body_len, r->write_ms, r->read_ms,
        r->parsed, r->mismatched_len, r->parse_failures,
        note ? note : "");
    bench_csv_append(csv_path, tag, "longmsg",
                     (uint64_t)r->bytes_written,
                     (uint64_t)n,
                     r->write_ms + r->read_ms,
                     r->peak_rss_kb,
                     full_note);
}

// ---------------------------------------------------------------------------
// Tests

#define KB (size_t)1024
#define MB (size_t)(1024 * 1024)

static void
run_simple(const char* tmp, const char* csv, const char* tag,
           int n, size_t body_len, pattern_t pat, const char* note)
{
    roundtrip_result_t r = { 0 };
    run_roundtrip(tmp, tag, n, body_len, pat, &r);
    fprintf(stderr,
            "  %-7s n=%d body=%zu B  write=%.1fms read=%.1fms parsed=%" PRId64
            " mismatch=%" PRId64 " fail=%" PRId64 "\n",
            tag, n, body_len, r.write_ms, r.read_ms,
            r.parsed, r.mismatched_len, r.parse_failures);
    csv_emit(csv, tag, body_len, n, &r, note);
    roundtrip_cleanup(&r);
}

// L7: body just over FF_MAX_LINE_LEN. ff_readline must reject and return "".
// We can't roundtrip via ff_write_line because the writer doesn't enforce a
// limit; we craft the line manually and verify the reader's rejection.
static void
run_oversized(const char* tmp, const char* csv)
{
    const char* tag = "L7";
    auto_gchar gchar* path = g_strdup_printf("%s/%s.log", tmp, tag);
    size_t body_len = FF_MAX_LINE_LEN + 1;

    FILE* fp = fopen(path, "w");
    if (!fp) {
        fprintf(stderr, "  %s: cannot open %s\n", tag, path);
        return;
    }
    fprintf(fp, "%s", FLATFILE_HEADER);
    // Hand-build a single oversized line: timestamp + meta + sender: + body + \n
    auto_gchar gchar* body = make_body(body_len, PAT_FILLER);
    fprintf(fp,
            "2025-06-15T12:00:00Z [chat|none|id:over] alice@bench/phone: %s\n",
            body);
    // Add one normal line afterwards so we can verify the loop continues.
    fprintf(fp, "2025-06-15T12:01:00Z [chat|none|id:next] alice@bench/phone: ok\n");
    fclose(fp);

    bench_drop_page_cache(path);
    double t0 = bench_now_ms();

    fp = fopen(path, "r");
    ff_skip_bom(fp);
    int64_t parsed = 0, empty_skipped = 0, failures = 0;
    char* buf;
    while ((buf = ff_readline(fp, NULL)) != NULL) {
        if (buf[0] == '\0') {
            empty_skipped++;
            free(buf);
            continue;
        }
        if (buf[0] == '#') {
            free(buf);
            continue;
        }
        ff_parsed_line_t* pl = ff_parse_line(buf);
        free(buf);
        if (pl) { parsed++; ff_parsed_line_free(pl); }
        else failures++;
    }
    fclose(fp);
    double read_ms = bench_now_ms() - t0;
    long rss = bench_peak_rss_kb();

    int ok = (parsed == 1 && empty_skipped >= 1);
    fprintf(stderr,
            "  L7      body=%zu  read=%.1fms  parsed=%" PRId64
            " skipped_empty=%" PRId64 " failures=%" PRId64 "  %s\n",
            body_len, read_ms, parsed, empty_skipped, failures,
            ok ? "OK (oversize rejected)" : "MISBEHAVE");
    if (csv) {
        struct stat st;
        uint64_t sz = (stat(path, &st) == 0) ? (uint64_t)st.st_size : 0;
        auto_gchar gchar* note = g_strdup_printf(
            "oversize=%zu_rejected=%s read=%.1fms parsed=%" PRId64,
            body_len, ok ? "yes" : "no", read_ms, parsed);
        bench_csv_append(csv, tag, "longmsg", sz, 2, read_ms, rss, note);
    }
    unlink(path);
}

// L12: pagination — last 100 messages where 5 of the last 100 are 1 MB bodies.
// Simulates a chat scrollback that includes a few paste-bombs near the end.
static void
run_pagination_with_huge(const char* tmp, const char* csv)
{
    const char* tag = "L12";
    auto_gchar gchar* path = g_strdup_printf("%s/%s.log", tmp, tag);

    int total = 1000;
    int huge_indices[5] = { 950, 970, 985, 992, 998 }; // five 1MB bodies near end

    FILE* fp = fopen(path, "w");
    if (!fp) return;
    fprintf(fp, "%s", FLATFILE_HEADER);
    GDateTime* base = g_date_time_new_utc(2025, 6, 15, 12, 0, 0.0);
    for (int i = 0; i < total; i++) {
        GDateTime* t = g_date_time_add_seconds(base, i);
        auto_gchar gchar* iso = g_date_time_format_iso8601(t);
        g_date_time_unref(t);
        size_t blen = 100;
        for (int k = 0; k < 5; k++) if (huge_indices[k] == i) { blen = MB; break; }
        auto_gchar gchar* sid = g_strdup_printf("page-%d", i);
        auto_gchar gchar* body = make_body(blen, PAT_FILLER);
        ff_write_line(fp, iso, "chat", "none",
                      sid, NULL, NULL,
                      "alice@bench.example", "phone",
                      NULL, NULL, -1,
                      body);
    }
    g_date_time_unref(base);
    fclose(fp);

    bench_drop_page_cache(path);
    double t0 = bench_now_ms();

    // Build state, find the second-to-last index entry to simulate "last 100".
    ff_contact_state_t* state = ff_state_new(path);
    ff_state_ensure_fresh(state);
    off_t start = state->bom_len;
    if (state->n_entries >= 2)
        start = state->entries[state->n_entries - 2].byte_offset;
    else if (state->n_entries == 1)
        start = state->entries[state->n_entries - 1].byte_offset;
    ff_state_free(state);

    fp = fopen(path, "r");
    fseeko(fp, start, SEEK_SET);
    char* ring[100] = { 0 };
    int rpos = 0;
    int64_t parsed = 0;
    char* buf;
    while ((buf = ff_readline(fp, NULL)) != NULL) {
        if (buf[0] == '\0' || buf[0] == '#') { free(buf); continue; }
        ff_parsed_line_t* pl = ff_parse_line(buf);
        free(buf);
        if (!pl) continue;
        parsed++;
        if (ring[rpos]) g_free(ring[rpos]);
        ring[rpos] = g_strdup(pl->message ? pl->message : "");
        rpos = (rpos + 1) % 100;
        ff_parsed_line_free(pl);
    }
    fclose(fp);
    double dt = bench_now_ms() - t0;
    long rss = bench_peak_rss_kb();
    for (int i = 0; i < 100; i++) g_free(ring[i]);

    fprintf(stderr,
            "  L12     scrollback w/ 5×1MB in last 100  read=%.1fms parsed=%" PRId64 " rss=%ldKB\n",
            dt, parsed, rss);
    if (csv) {
        struct stat st;
        uint64_t sz = (stat(path, &st) == 0) ? (uint64_t)st.st_size : 0;
        auto_gchar gchar* note = g_strdup_printf(
            "5x1MB_in_last_100 parsed=%" PRId64, parsed);
        bench_csv_append(csv, tag, "longmsg", sz, (uint64_t)parsed, dt, rss, note);
    }
    unlink(path);
}

// ---------------------------------------------------------------------------
// Driver

int
main(int argc, char** argv)
{
    cli_t cli;
    if (parse_args(argc, argv, &cli) != 0)
        return 2;

    if (g_mkdir_with_parents(cli.tmp_dir, 0755) != 0) {
        fprintf(stderr, "cannot create %s\n", cli.tmp_dir);
        return 2;
    }

    fprintf(stderr, "===== bench_long_messages: tmp=%s =====\n", cli.tmp_dir);

    if (test_enabled(cli.tests, "L1"))
        run_simple(cli.tmp_dir, cli.csv_path, "L1", 100, 1 * KB, PAT_FILLER, "1KB body x100");
    if (test_enabled(cli.tests, "L2"))
        run_simple(cli.tmp_dir, cli.csv_path, "L2", 100, 10 * KB, PAT_FILLER, "10KB body x100");
    if (test_enabled(cli.tests, "L3"))
        run_simple(cli.tmp_dir, cli.csv_path, "L3", 100, 100 * KB, PAT_FILLER, "100KB body x100");
    if (test_enabled(cli.tests, "L4"))
        run_simple(cli.tmp_dir, cli.csv_path, "L4", 50, 1 * MB, PAT_FILLER, "1MB body x50");
    if (test_enabled(cli.tests, "L5"))
        run_simple(cli.tmp_dir, cli.csv_path, "L5", 10, 5 * MB, PAT_FILLER, "5MB body x10");
    if (test_enabled(cli.tests, "L6"))
        run_simple(cli.tmp_dir, cli.csv_path, "L6", 4, 9 * MB + 900 * KB, PAT_FILLER,
                   "9.9MB body x4 (just under FF_MAX_LINE_LEN)");
    if (test_enabled(cli.tests, "L7"))
        run_oversized(cli.tmp_dir, cli.csv_path);
    if (test_enabled(cli.tests, "L8"))
        run_simple(cli.tmp_dir, cli.csv_path, "L8", 50, 100 * KB, PAT_EMBEDDED_LF,
                   "100KB body w/ \\n every 100B");
    if (test_enabled(cli.tests, "L9"))
        run_simple(cli.tmp_dir, cli.csv_path, "L9", 50, 100 * KB, PAT_EMBEDDED_PIPE,
                   "100KB body w/ pipes");
    if (test_enabled(cli.tests, "L10"))
        run_simple(cli.tmp_dir, cli.csv_path, "L10", 50, 100 * KB, PAT_EMOJI,
                   "100KB body utf-8 emoji");
    if (test_enabled(cli.tests, "L11"))
        run_simple(cli.tmp_dir, cli.csv_path, "L11", 100, 1 * KB, PAT_FILLER, "sanity baseline");
    if (test_enabled(cli.tests, "L12"))
        run_pagination_with_huge(cli.tmp_dir, cli.csv_path);
    if (test_enabled(cli.tests, "L13"))
        run_simple(cli.tmp_dir, cli.csv_path, "L13", 100, 1 * MB, PAT_FILLER,
                   "verify-equiv full parse on 100x1MB");
    if (test_enabled(cli.tests, "L14"))
        run_simple(cli.tmp_dir, cli.csv_path, "L14", 1000, 100 * KB, PAT_FILLER,
                   "rapid append 1000x100KB");

    return 0;
}