rayforce-sys 1.2.0

/*
 *   Copyright (c) 2025-2026 Anton Kundenko <singaraiona@gmail.com>
 *   All rights reserved.

 *   Permission is hereby granted, free of charge, to any person obtaining a copy
 *   of this software and associated documentation files (the "Software"), to deal
 *   in the Software without restriction, including without limitation the rights
 *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *   copies of the Software, and to permit persons to whom the Software is
 *   furnished to do so, subject to the following conditions:

 *   The above copyright notice and this permission notice shall be included in all
 *   copies or substantial portions of the Software.

 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 *   SOFTWARE.
 */

#define _POSIX_C_SOURCE 200809L
#include "core/poll.h"
#include "core/ipc.h"
#include "core/pool.h"
#include "core/profile.h"
#include "core/qlog.h"
#include "core/crash.h"
#include "app/repl.h"
#include "core/runtime.h"
#include "store/journal.h"
#include "ops/internal.h"
#include "ops/idxop.h"
#include "mem/heap.h"
#include <rayforce.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <signal.h>
#include <unistd.h>

/* Parse a byte size with an optional K/M/G/T suffix (base-1024).  Returns
 * the byte count, or -1 on a malformed value.  Used by the -m/--mem RAM
 * budget flag: allocations that would push the anonymous (RAM-resident)
 * footprint past this bound spill to a disk-backed mapping instead of
 * OOM-ing (see ray_heap_set_anon_watermark). */
static int64_t parse_size_arg(const char* s) {
    if (!s || !*s) return -1;
    char* end = NULL;
    errno = 0;
    double v = strtod(s, &end);
    if (errno != 0 || end == s || v < 0) return -1;
    int64_t mult = 1;
    if (*end) {
        switch (*end) {
            case 'k': case 'K': mult = 1LL << 10; break;
            case 'm': case 'M': mult = 1LL << 20; break;
            case 'g': case 'G': mult = 1LL << 30; break;
            case 't': case 'T': mult = 1LL << 40; break;
            default: return -1;
        }
        if (end[1] != '\0') return -1;   /* trailing junk after the suffix */
    }
    return (int64_t)(v * (double)mult);
}

/* --------------------------------------------------------------------------
 * Flag-value guard (#600)
 *
 * Every value-taking flag used to be gated on `i + 1 < argc` and then
 * consume argv[++i] blindly, so one flag silently ate the next: `-Q -p
 * 5099` started a server with NO listener, no error and no warning —
 * the worst shape for a service, because the unit looks healthy while
 * every client gets connection refused.  A trailing flag fell through to
 * the positional-file arm and reported `cannot open '-Q'`.
 *
 * A value is refused when it is missing, or when it is EXACTLY one of
 * this program's flag tokens.  A value that merely STARTS with '-' (a
 * password, a negative number) stays legal — refusing those would break
 * working command lines for no gain.
 *
 * Keep this list in step with the flags handled in main's parse loop and
 * with --help.  (ray_build_sys_args in src/ops/system.c walks argv a
 * second time for .sys.args; it only ever sees a command line that
 * already passed this guard, so it cannot mis-pair.)
 * -------------------------------------------------------------------------- */
static const char* const k_known_flags[] = {
    "-i", "--interactive", "-p", "--port", "-c", "--cores",
    "-t", "--timeit", "-Q", "--querylog", "-f", "--file",
    "-u", "-U", "-l", "-L", "-m", "--mem", "-h", "--help",
    "--",   /* the app-args terminator is not a value either */
    NULL
};

static bool arg_is_known_flag(const char* tok) {
    for (int k = 0; k_known_flags[k]; k++)
        if (strcmp(tok, k_known_flags[k]) == 0) return true;
    return false;
}

/* The value for a value-taking flag, or NULL with the diagnostic already
 * printed.  Advances *i past the value it returns. */
static const char* flag_value(int argc, char** argv, int* i, const char* name) {
    if (*i + 1 >= argc) {
        fprintf(stderr, "error: %s expects a value\n", name);
        return NULL;
    }
    const char* v = argv[*i + 1];
    if (arg_is_known_flag(v)) {
        fprintf(stderr, "error: %s expects a value, got flag '%s'\n", name, v);
        return NULL;
    }
    ++*i;
    return v;
}

int main(int argc, char** argv) {
    /* Install the fatal-signal handler first: a crash during startup
     * should still produce a symbolized backtrace.  Handles the fault
     * signals only; term.c owns the graceful-shutdown set. */
    ray_crash_install();
    ray_expr_stats_init();
    ray_idx_stats_init();
    ray_runtime_t* rt = ray_runtime_create(argc, argv);
    if (!rt) { fprintf(stderr, "failed to create runtime\n"); return 1; }

    int rc = 0;
    int interactive = 0;
    const char* file = NULL;
    uint16_t port = 0;
    char bind_host[64] = "";   /* -p HOST:PORT bind address; empty = all interfaces */
    int n_cores = -1;      /* total execution cores; -1 = leave pool lazy */
    int timeit_init = 0;   /* -t N: enable profiler at startup */
    int qlog_init = 0;     /* -Q N: enable query-statistics logging at startup */
    const char* auth_pw = NULL;
    bool auth_restricted = false;
    const char* log_base = NULL;
    ray_journal_mode_t log_mode = RAY_JOURNAL_OFF;

    /* Parse args. Supported flags:
     *   -f FILE          run script file
     *   -p PORT|HOST:PORT  IPC listen port, optionally bound to HOST
     *   -c N             total execution cores, main included (0 = auto)
     *   -t N             enable timeit at startup (N != 0 turns on)
     *   -Q N             enable query-statistics logging (N != 0 turns on)
     *   -i               interactive
     * Plus rayforce-specific:
     *   -u PW / -U PW    auth password (plain / restricted)
     *   -h               help
     */
    for (int i = 1; i < argc; i++) {
        if (strcmp(argv[i], "-i") == 0 || strcmp(argv[i], "--interactive") == 0)
            interactive = 1;
        else if (strcmp(argv[i], "-p") == 0 || strcmp(argv[i], "--port") == 0) {
            /* PORT binds all interfaces; HOST:PORT confines the listener
             * to one IPv4 address (#427).  Strict digits-only port — a
             * silently truncated "65536x" listener is worse than exiting. */
            const char* parg = flag_value(argc, argv, &i, argv[i]);
            if (!parg) { ray_runtime_destroy(rt); return 2; }
            const char* colon = strchr(parg, ':');
            const char* pstr  = parg;
            if (colon) {
                size_t hlen = (size_t)(colon - parg);
                if (hlen == 0 || hlen >= sizeof(bind_host)) {
                    fprintf(stderr, "bad -p bind address: %s\n", parg);
                    ray_runtime_destroy(rt);
                    return 2;
                }
                memcpy(bind_host, parg, hlen);
                bind_host[hlen] = '\0';
                pstr = colon + 1;
            }
            char* endp;
            errno = 0;
            long pv = strtol(pstr, &endp, 10);
            if (endp == pstr || *endp != '\0' || errno == ERANGE || pv <= 0 || pv > 65535) {
                fprintf(stderr, "bad -p port (must be 1..65535): %s\n", parg);
                ray_runtime_destroy(rt);
                return 2;
            }
            port = (uint16_t)pv;
        }
        else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "--cores") == 0) {
            const char* v = flag_value(argc, argv, &i, argv[i]);
            if (!v) { ray_runtime_destroy(rt); return 2; }
            n_cores = atoi(v);
            if (n_cores < 0) n_cores = 0;
        }
        else if (strcmp(argv[i], "-t") == 0 || strcmp(argv[i], "--timeit") == 0) {
            const char* v = flag_value(argc, argv, &i, argv[i]);
            if (!v) { ray_runtime_destroy(rt); return 2; }
            timeit_init = (atoi(v) != 0);
        }
        else if (strcmp(argv[i], "-Q") == 0 || strcmp(argv[i], "--querylog") == 0) {
            const char* v = flag_value(argc, argv, &i, argv[i]);
            if (!v) { ray_runtime_destroy(rt); return 2; }
            qlog_init = (atoi(v) != 0);
        }
        else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--file") == 0) {
            file = flag_value(argc, argv, &i, argv[i]);
            if (!file) { ray_runtime_destroy(rt); return 2; }
        }
        else if (strcmp(argv[i], "-u") == 0 || strcmp(argv[i], "-U") == 0) {
            auth_restricted = (argv[i][1] == 'U');
            auth_pw = flag_value(argc, argv, &i, argv[i]);
            if (!auth_pw) { ray_runtime_destroy(rt); return 2; }
        }
        else if (strcmp(argv[i], "-l") == 0 || strcmp(argv[i], "-L") == 0) {
            log_mode = (argv[i][1] == 'L') ? RAY_JOURNAL_SYNC : RAY_JOURNAL_ASYNC;
            log_base = flag_value(argc, argv, &i, argv[i]);
            if (!log_base) { ray_runtime_destroy(rt); return 2; }
        }
        else if (strcmp(argv[i], "-m") == 0 || strcmp(argv[i], "--mem") == 0) {
            const char* mv = flag_value(argc, argv, &i, argv[i]);
            if (!mv) { ray_runtime_destroy(rt); return 2; }
            int64_t wm = parse_size_arg(mv);
            if (wm <= 0) {
                fprintf(stderr, "error: -m/--mem expects a size like 4G, 512M, or a byte count\n");
                ray_runtime_destroy(rt);
                return 1;
            }
            ray_heap_set_anon_watermark(wm);
        }
        else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
            fprintf(stdout,
                "usage: %s [-f file] [-p port] [-c cores] [-t 0|1] [-Q 0|1] [-i]"
                " [-u PW | -U PW] [-l BASE | -L BASE] [-m SIZE] [file.rfl] [-- app args]\n"
                "  -f, --file FILE     run script file (or pass as a positional arg)\n"
                "  -p, --port PORT     listen for IPC clients on PORT (all interfaces)\n"
                "      --port HOST:PORT  bind the listener to HOST only (e.g. 127.0.0.1:7701)\n"
                "  -c, --cores N       total execution cores, main included (0 = auto)\n"
                "  -t, --timeit N      enable profiler at startup (N != 0)\n"
                "  -Q, --querylog N    enable query logging at startup (N != 0);\n"
                "                      read the ring with (.sys.querylog)\n"
                "  -i, --interactive   start the REPL even after running a file\n"
                "  -u PW               set plain auth password\n"
                "  -U PW               set restricted auth password\n"
                "  -l BASE             enable journal logging (BASE.log + BASE.qdb)\n"
                "  -L BASE             enable journal logging with fsync per write\n"
                "  -m, --mem SIZE      RAM budget (e.g. 4G, 512M); allocations past it\n"
                "                      spill to disk instead of OOM (default: physical RAM)\n"
                "  -h, --help          show this message\n"
                "  --                  pass following args to the app; read\n"
                "                      them in Rayfall via (.sys.args)\n"
                "\nFor a remote REPL, start rayforce locally and call\n"
                "(.repl.connect \"host:port\") inside the REPL.\n"
                "\nRunning as a service:\n"
                "  When stdin and stdout are both not a terminal (systemd\n"
                "  Type=simple, Docker, nohup &, etc.), rayforce skips the\n"
                "  REPL and runs only the IPC poll loop.  Redirect stdout\n"
                "  and stderr to a file/journal — anything (println ...)\n"
                "  prints from the server side goes there.  Example:\n"
                "    nohup rayforce -p 5000 > rayforce.log 2>&1 &\n"
                "    systemd:  StandardOutput=journal StandardError=journal\n"
                "  With -p and a script, a script that fails to parse or\n"
                "  evaluate is fatal: the listener closes and the process\n"
                "  exits non-zero (with -i the REPL stays up instead).\n",
                argv[0]);
            ray_runtime_destroy(rt);
            return 0;
        }
        else if (strcmp(argv[i], "--") == 0)
            break;   /* stop flag parsing; remaining tokens are app args */
        else if (argv[i][0] == '-' && argv[i][1] != '\0') {
            /* Not a known flag.  This used to fall into the positional arm
             * below, so a typo'd option was silently swallowed as the script
             * name and then overwritten by the real one — the flag simply
             * did nothing (#600).  A lone "-" stays a positional. */
            fprintf(stderr, "error: unknown option '%s'\n", argv[i]);
            ray_runtime_destroy(rt);
            return 2;
        }
        else
            file = argv[i];
    }

    /* Expose the full command line to Rayfall via (.sys.args). */
    ray_t* sys_args = ray_build_sys_args(argc, argv);
    if (!sys_args || RAY_IS_ERR(sys_args)) {
        fprintf(stderr, "error: invalid command-line arguments\n");
        if (sys_args && RAY_IS_ERR(sys_args)) ray_error_free(sys_args);
        ray_runtime_destroy(rt);
        return 2;
    }
    ray_runtime_set_sys_args(sys_args);

    /* Initialise the worker pool before anything else that might use it
     * (file load, REPL eval, builtins).  If -c wasn't given, leave the
     * pool to its lazy default on first use. */
    if (n_cores >= 0) {
        ray_err_t err = ray_pool_init_total((uint32_t)n_cores);
        if (err != RAY_OK)
            fprintf(stderr, "warning: ray_pool_init_total(%d) failed (%d)\n",
                    n_cores, (int)err);
    }

    /* -t at startup flips the global profiler flag. The REPL reads
     * g_ray_profile.active into repl->timeit on create. */
    if (timeit_init)
        g_ray_profile.active = true;

    /* -Q at startup arms the ambient query-statistics ring (.sys.querylog). */
    if (qlog_init)
        ray_qlog_set_enabled(true);

    ray_poll_t* poll = ray_poll_create();
    /* Publish the main poll into the runtime so runtime-level builtins
     * (`.sys.cmd "listen N"`, `.sys.listen N`) can attach listeners to
     * the same event loop the REPL/server is driving. */
    if (poll) ray_runtime_set_poll(poll);

    if (poll && auth_pw) {
        size_t pw_len = strlen(auth_pw);
        if (pw_len >= sizeof(poll->auth_secret))
            pw_len = sizeof(poll->auth_secret) - 1;
        memcpy(poll->auth_secret, auth_pw, pw_len);
        poll->auth_secret[pw_len] = '\0';
        ray_poll_set_restricted(poll, auth_restricted);
    }

    /* Open journal BEFORE the IPC listener accepts any connection.
     * Replay must complete with a stable global env, and any badtail
     * is fatal — exit non-zero so a supervisor can flag the failure
     * rather than silently coming up with a partially-applied state. */
    if (log_base) {
        ray_err_t le = ray_journal_open(log_base, log_mode);
        if (le != RAY_OK) {
            fprintf(stderr, "log: open failed for base=%s (rc=%d)\n",
                    log_base, (int)le);
            if (poll) ray_poll_destroy(poll);
            ray_runtime_destroy(rt);
            return 2;
        }
    }

    /* Start IPC server if port specified */
    if (port > 0) {
        const char* bh = bind_host[0] ? bind_host : NULL;
        if (poll && ray_ipc_listen_at(poll, bh, port) >= 0) {
            fprintf(stderr, "listening on %s:%u\n", bh ? bh : "*", port);
        } else {
            /* The process was asked to serve.  Carrying on would run the
             * script and exit 0 (or sit on its timers) with no listener,
             * and a supervisor would learn of it from the first client's
             * ECONNREFUSED.  Fail like every other server does. */
            fprintf(stderr, "failed to listen on %s:%u: %s\n",
                    bh ? bh : "*", port, strerror(errno));
            rc = 1;
            goto done;
        }
    }

    /* Load script if specified */
    if (file) {
        rc = ray_repl_run_file(file);
        /* The listener is already up (so the script may use it), which
         * means a failed startup script would otherwise leave a port
         * that accepts connections in front of an empty workspace: a
         * supervisor's port check passes and Restart=on-failure never
         * fires.  Non-interactively, a script that fails to parse or
         * evaluate is fatal; the listener goes down with the process.
         * With -i the REPL stays up for the user to look around. */
        if (rc != 0 && !interactive && port > 0) {
            fprintf(stderr, "startup script %s failed; closing listener on %s:%u and exiting\n",
                    file, bind_host[0] ? bind_host : "*", port);
            goto done;
        }
        if (!interactive && !(port > 0)) {
            /* A script that armed timers means to run them: stay until
             * they are spent (a one-shot fires once, a periodic one until
             * the script ends the process).  Nothing else keeps a script
             * alive — a client handle it left open is not a reason. */
            if (poll) ray_poll_drain_timers(poll);
            goto done;
        }
    }

    /* REPL or pure server mode.
     *
     * Auto-detect server-only when launched as a daemon-style service:
     * `-p` set, no script, no `-i`, and neither stdin nor stdout is a
     * terminal.  In that case we skip ray_repl_create entirely and just
     * drive the poll loop — which is the right shape for `nohup ... &`,
     * systemd `Type=simple`, Docker without `-it`, etc.
     *
     * Without this guard the local REPL would enter run_piped() and
     * block on fgets, starving the same poll loop the IPC listener is
     * registered on — so connections get accepted by the kernel but
     * never user-space handled.  Auto-detect fixes the deploy hole;
     * `-i` overrides for the rare case of "I want REPL even with stdin
     * piped". */
    bool stdin_tty  = isatty(STDIN_FILENO);
    bool stdout_tty = isatty(STDOUT_FILENO);
    bool server_only = poll && port > 0 && !file && !interactive
                       && !stdin_tty && !stdout_tty;

    if (server_only) {
        /* In service deployments stdout/stderr are typically pointed
         * at a journal pipe or log file by the supervisor (systemd,
         * docker, nohup).  Anything (println …) prints from the
         * server side ends up there — see --help "Running as a
         * service" for guidance.  We deliberately don't auto-create
         * a log file: that would conflict with whatever the admin
         * already configured. */
        /* Suppress SIGPIPE so a write to a stdout/stderr pipe whose
         * peer has gone away (journald restart, log rotator, etc.)
         * doesn't kill the daemon.  Writes still fail with EPIPE,
         * libc sets ferror(stdout), and the bytes are simply lost —
         * which is the right behaviour for a service whose log sink
         * is recoverable.  Only set in server-only mode; an
         * interactive REPL with a broken pipe is genuinely fatal. */
#ifndef RAY_OS_WINDOWS
        signal(SIGPIPE, SIG_IGN);
#endif
        fprintf(stderr, "no terminal — running in server-only mode\n");
        ray_poll_run(poll);
    } else {
        ray_repl_t* repl = ray_repl_create(poll);
        if (repl) {
            ray_repl_run(repl);
            ray_repl_destroy(repl);
        } else if (poll && port > 0) {
            /* REPL create failed (OOM etc.) — fall back to pure poll. */
            ray_poll_run(poll);
        }
    }

done:
    /* Flush + close the journal before the rest of the runtime tears
     * down — fclose triggers a final fflush, which is what `-l` mode
     * relies on for "best-effort" durability at clean shutdown. */
    ray_journal_close();
    if (poll) ray_poll_destroy(poll);
    ray_runtime_destroy(rt);
    return rc;
}