File Coverage

MariaDB.xs

Criterion	Covered	Total	%
statement	104	1561	6.6
branch	37	1196	3.0
condition			n/a
subroutine			n/a
pod			n/a
total	141	2757	5.1

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.h"

#include "EVAPI.h"
#include

#include "ngx_queue.h"

typedef struct ev_mariadb_s ev_mariadb_t;
typedef struct ev_mariadb_cb_s ev_mariadb_cb_t;
typedef struct ev_mariadb_send_s ev_mariadb_send_t;
typedef struct ev_mariadb_stmt_s ev_mariadb_stmt_t;

typedef ev_mariadb_t* EV__MariaDB;
typedef struct ev_loop* EV__Loop;

#define EV_MARIADB_MAGIC 0xDEADBEEF
#define EV_MARIADB_FREED 0xFEEDFACE

enum ev_mariadb_state {
    STATE_IDLE,
    STATE_CONNECTING,
    STATE_SEND,
    STATE_READ_RESULT,
    STATE_STORE_RESULT,
    STATE_NEXT_RESULT,
    STATE_PING,
    STATE_CHANGE_USER,
    STATE_SELECT_DB,
    STATE_RESET_CONNECTION,
    STATE_SET_CHARSET,
    STATE_COMMIT,
    STATE_ROLLBACK,
    STATE_AUTOCOMMIT,
    STATE_STMT_PREPARE,     /* boundary: states >= here block query queueing */
    STATE_STMT_EXECUTE,
    STATE_STMT_STORE,
    STATE_STMT_CLOSE,
    STATE_STMT_RESET,
    STATE_STMT_SEND_LONG_DATA,
    STATE_REAL_QUERY,
    STATE_STREAM_FETCH,
    STATE_CLOSE,
};

struct ev_mariadb_s {
    unsigned int magic;
    struct ev_loop *loop;
    MYSQL *conn;

    ev_io    rio, wio;
    ev_timer timer;
    int      reading, writing, timing;
    int      fd;

    enum ev_mariadb_state state;
    char    *host, *user, *password, *database, *unix_socket;
    unsigned int port;

    ngx_queue_t cb_queue;       /* callbacks for sent queries (awaiting results) */
    ngx_queue_t send_queue;     /* queries waiting to be sent */
    int         pending_count;  /* total: send_queue + cb_queue */
    int         send_count;     /* queries sent, results not yet read */
    int         draining;       /* draining multi-result extras */

    /* current operation context */
    int          op_ret;
    MYSQL_RES   *op_result;
    MYSQL_STMT  *op_stmt;
    ev_mariadb_stmt_t *op_stmt_ctx;  /* per-stmt wrapper for bind_params cleanup */
    ev_mariadb_stmt_t *stmt_list;    /* all allocated stmt wrappers */
    MYSQL       *op_conn_ret;
    my_bool      op_bool_ret;
    MYSQL_ROW    op_row;        /* for streaming fetch_row result */
    SV          *stream_cb;     /* streaming per-row callback */
    char        *op_data_ptr;   /* copied data buffer for send_long_data */

    SV *on_connect;
    SV *on_error;

    int callback_depth;
    pid_t connect_pid;      /* PID at connect time, for fork detection */

    /* connection options (applied before mysql_real_connect_start) */
    unsigned int connect_timeout;
    unsigned int read_timeout;
    unsigned int write_timeout;
    int          compress;
    int          multi_statements;
    int          found_rows;
    char        *charset;
    char        *init_command;
    char        *ssl_key;
    char        *ssl_cert;
    char        *ssl_ca;
    char        *ssl_capath;
    char        *ssl_cipher;
    int          ssl_verify_server_cert;
    int          utf8;  /* auto-flag result strings as UTF-8 */
    unsigned long client_flags;

    /* Pending cache values for in-flight change_user/select_db/set_charset.
       Committed by on_utility_done on success, freed on failure. The strings
       are also passed to the libmariadb start function so they remain valid
       across the async op. */
    char *pending_user;
    char *pending_password;
    char *pending_database;
    char *pending_charset;
};

struct ev_mariadb_cb_s {
    SV          *cb;
    ngx_queue_t  queue;
};

struct ev_mariadb_send_s {
    char        *sql;
    unsigned long sql_len;
    SV          *cb;
    ngx_queue_t  queue;
};

struct ev_mariadb_stmt_s {
    MYSQL_STMT  *stmt;
    MYSQL_BIND  *bind_params;
    int          bind_param_count;
    int          closed;      /* invalidated by cleanup_connection */
    ev_mariadb_stmt_t *next;  /* linked list of all stmts on this connection */
};

#define MAX_PIPELINE_DEPTH 64
#define MAX_FREELIST_DEPTH 64

#define COPY_ERROR(buf, src) do { \
    strncpy((buf), (src), sizeof(buf) - 1); \
    (buf)[sizeof(buf) - 1] = '\0'; \
} while (0)

#define SET_STR_OPTION(field) do { \
    if (self->field) Safefree(self->field); \
    self->field = SvOK(value) ? safe_strdup(SvPV_nolen(value)) : NULL; \
} while (0)

#define IS_FORKED(self) \
    ((self)->connect_pid != 0 && (self)->connect_pid != getpid())

/* Safefree is NULL-safe; the NULL assignment matters for callers where
   self survives the call (cleanup_connection, reconnect via reset()). */
#define FREE_STR(p) do { Safefree(p); (p) = NULL; } while (0)

#define CHECK_READY(self, cb) \
    do { \
        if (NULL == (self)->conn || (self)->state == STATE_CONNECTING) \
            croak("not connected"); \
        if (IS_FORKED(self)) \
            croak("connection not valid after fork"); \
        if ((self)->state != STATE_IDLE) \
            croak("another operation is in progress"); \
        if ((self)->send_count > 0) \
            croak("cannot start operation while pipeline results are pending"); \
        if (!(SvROK(cb) && SvTYPE(SvRV(cb)) == SVt_PVCV)) \
            croak("callback must be a CODE reference"); \
    } while (0)

#define CHECK_STMT(ctx) \
    do { \
        if (!(ctx)) \
            croak("invalid statement handle"); \
        if ((ctx)->closed) \
            croak("statement handle is no longer valid (connection was reset)"); \
    } while (0)

static void io_cb(EV_P_ ev_io *w, int revents);
static void timer_cb(EV_P_ ev_timer *w, int revents);
static void continue_operation(ev_mariadb_t *self, int events);
static void pipeline_advance(ev_mariadb_t *self);
static void on_next_result_done(ev_mariadb_t *self);
static void start_reading(ev_mariadb_t *self);
static void stop_reading(ev_mariadb_t *self);
static void start_writing(ev_mariadb_t *self);
static void stop_writing(ev_mariadb_t *self);
static void free_stmt_bind_params(ev_mariadb_stmt_t *ctx);
static int  is_utf8_charset(unsigned int charsetnr);
static void start_timer(ev_mariadb_t *self);
static void stop_timer(ev_mariadb_t *self);
static void update_watchers(ev_mariadb_t *self, int status);
static void emit_error(ev_mariadb_t *self, const char *msg);
static void cleanup_connection(ev_mariadb_t *self);
static int  check_destroyed(ev_mariadb_t *self);
static void drain_multi_result(ev_mariadb_t *self);
static AV*  build_field_names(MYSQL *conn, MYSQL_FIELD *fields, unsigned int ncols);
static void on_real_query_done(ev_mariadb_t *self);
static void on_stream_fetch_done(ev_mariadb_t *self);
static void on_close_done(ev_mariadb_t *self);

static void maybe_pipeline(ev_mariadb_t *self) {
    if (self->state != STATE_IDLE) return;
    if (!ngx_queue_empty(&self->send_queue)) {
        pipeline_advance(self);
        return;
    }
    stop_reading(self);
    stop_writing(self);
    stop_timer(self);
}

/* --- freelist for cb_queue entries --- */

static ev_mariadb_cb_t *cbt_freelist = NULL;
static int cbt_freelist_size = 0;

static ev_mariadb_cb_t* alloc_cbt(void) {
    ev_mariadb_cb_t *cbt;
    if (cbt_freelist) {
        cbt = cbt_freelist;
        cbt_freelist = *(ev_mariadb_cb_t **)cbt;
        cbt_freelist_size--;
    } else {
        Newx(cbt, 1, ev_mariadb_cb_t);
    }
    return cbt;
}

static void release_cbt(ev_mariadb_cb_t *cbt) {
    if (cbt_freelist_size >= MAX_FREELIST_DEPTH) {
        Safefree(cbt);
        return;
    }
    *(ev_mariadb_cb_t **)cbt = cbt_freelist;
    cbt_freelist = cbt;
    cbt_freelist_size++;
}

/* --- freelist for send_queue entries --- */

static ev_mariadb_send_t *send_freelist = NULL;
static int send_freelist_size = 0;

static ev_mariadb_send_t* alloc_send(void) {
    ev_mariadb_send_t *s;
    if (send_freelist) {
        s = send_freelist;
        send_freelist = *(ev_mariadb_send_t **)s;
        send_freelist_size--;
    } else {
        Newx(s, 1, ev_mariadb_send_t);
    }
    return s;
}

static void release_send(ev_mariadb_send_t *s) {
    if (send_freelist_size >= MAX_FREELIST_DEPTH) {
        Safefree(s);
        return;
    }
    *(ev_mariadb_send_t **)s = send_freelist;
    send_freelist = s;
    send_freelist_size++;
}

/* Drain freelists at interpreter shutdown so Valgrind reports zero leaks. */
static void drain_freelists(pTHX_ void *unused) {
    PERL_UNUSED_VAR(unused);
    while (cbt_freelist) {
        ev_mariadb_cb_t *next = *(ev_mariadb_cb_t **)cbt_freelist;
        Safefree(cbt_freelist);
        cbt_freelist = next;
    }
    cbt_freelist_size = 0;
    while (send_freelist) {
        ev_mariadb_send_t *next = *(ev_mariadb_send_t **)send_freelist;
        Safefree(send_freelist);
        send_freelist = next;
    }
    send_freelist_size = 0;
}

static void push_send(ev_mariadb_t *self, const char *sql, STRLEN sql_len, SV *cb) {
    char *sql_copy;
    ev_mariadb_send_t *s;
    Newx(sql_copy, sql_len + 1, char);
    Copy(sql, sql_copy, sql_len + 1, char);

    s = alloc_send();
    s->sql = sql_copy;
    s->sql_len = (unsigned long)sql_len;
    s->cb = SvREFCNT_inc(cb);
    ngx_queue_insert_tail(&self->send_queue, &s->queue);
    self->pending_count++;
}

static void drain_queues_silent(ev_mariadb_t *self) {
    while (!ngx_queue_empty(&self->send_queue)) {
        ngx_queue_t *q = ngx_queue_head(&self->send_queue);
        ev_mariadb_send_t *send = ngx_queue_data(q, ev_mariadb_send_t, queue);
        ngx_queue_remove(q);
        Safefree(send->sql);
        SvREFCNT_dec(send->cb);
        release_send(send);
        self->pending_count--;
    }
    while (!ngx_queue_empty(&self->cb_queue)) {
        ngx_queue_t *q = ngx_queue_head(&self->cb_queue);
        ev_mariadb_cb_t *cbt = ngx_queue_data(q, ev_mariadb_cb_t, queue);
        ngx_queue_remove(q);
        SvREFCNT_dec(cbt->cb);
        release_cbt(cbt);
        self->pending_count--;
    }
}

/* --- watcher helpers --- */

static void start_reading(ev_mariadb_t *self) {
    if (!self->reading && self->fd >= 0) {
        ev_io_start(self->loop, &self->rio);
        self->reading = 1;
    }
}

static void stop_reading(ev_mariadb_t *self) {
    if (self->reading) {
        ev_io_stop(self->loop, &self->rio);
        self->reading = 0;
    }
}

static void start_writing(ev_mariadb_t *self) {
    if (!self->writing && self->fd >= 0) {
        ev_io_start(self->loop, &self->wio);
        self->writing = 1;
    }
}

static void stop_writing(ev_mariadb_t *self) {
    if (self->writing) {
        ev_io_stop(self->loop, &self->wio);
        self->writing = 0;
    }
}

static void start_timer(ev_mariadb_t *self) {
    /* libmariadb may report a fresh timeout on each _cont; always re-arm.
       ms == 0 means "fire as soon as possible" — use 1ms so the timer fires
       on the next loop iteration rather than stalling. */
    unsigned int ms = mysql_get_timeout_value_ms(self->conn);
    if (ms == 0) ms = 1;
    if (self->timing) ev_timer_stop(self->loop, &self->timer);
    ev_timer_set(&self->timer, ms / 1000.0, 0.0);
    ev_timer_start(self->loop, &self->timer);
    self->timing = 1;
}

static void stop_timer(ev_mariadb_t *self) {
    if (self->timing) {
        ev_timer_stop(self->loop, &self->timer);
        self->timing = 0;
    }
}

static void update_watchers(ev_mariadb_t *self, int status) {
    if (status & (MYSQL_WAIT_READ | MYSQL_WAIT_EXCEPT)) start_reading(self); else stop_reading(self);
    if (status & MYSQL_WAIT_WRITE) start_writing(self); else stop_writing(self);
    if (status & MYSQL_WAIT_TIMEOUT) start_timer(self); else stop_timer(self);
}

static void init_io_watchers(ev_mariadb_t *self) {
    ev_io_init(&self->rio, io_cb, self->fd, EV_READ);
    self->rio.data = (void *)self;
    ev_io_init(&self->wio, io_cb, self->fd, EV_WRITE);
    self->wio.data = (void *)self;
}

static int check_destroyed(ev_mariadb_t *self) {
    if (self->magic == EV_MARIADB_FREED &&
        self->callback_depth == 0) {
        Safefree(self);
        return 1;
    }
    return 0;
}

static void emit_error(ev_mariadb_t *self, const char *msg) {
    SV *cb;
    if (NULL == self->on_error) return;
    cb = sv_2mortal(SvREFCNT_inc_simple_NN(self->on_error));

    dSP;
    ENTER;
    SAVETMPS;
    PUSHMARK(SP);
    XPUSHs(sv_2mortal(newSVpv(msg, 0)));
    PUTBACK;

    call_sv(cb, G_DISCARD | G_EVAL);
    if (SvTRUE(ERRSV)) {
        warn("EV::MariaDB: exception in error handler: %s", SvPV_nolen(ERRSV));
        sv_setpvn(ERRSV, "", 0);
    }

    FREETMPS;
    LEAVE;
}

/* Pop and return the head callback from cb_queue. Caller must SvREFCNT_dec. */
static SV* pop_cb(ev_mariadb_t *self) {
    ngx_queue_t *q;
    ev_mariadb_cb_t *cbt;
    SV *cb;

    if (ngx_queue_empty(&self->cb_queue)) return NULL;

    q = ngx_queue_head(&self->cb_queue);
    cbt = ngx_queue_data(q, ev_mariadb_cb_t, queue);

    cb = cbt->cb;
    ngx_queue_remove(q);
    self->pending_count--;
    release_cbt(cbt);

    return cb;
}

/* Invoke a callback with (undef, errmsg). Decrements cb refcount. */
static void invoke_error_cb(SV *cb, const char *errmsg) {
    dSP;
    ENTER;
    SAVETMPS;
    PUSHMARK(SP);
    PUSHs(&PL_sv_undef);
    PUSHs(sv_2mortal(newSVpv(errmsg, 0)));
    PUTBACK;
    call_sv(cb, G_DISCARD | G_EVAL);
    if (SvTRUE(ERRSV)) {
        warn("EV::MariaDB: exception in callback: %s", SvPV_nolen(ERRSV));
        sv_setpvn(ERRSV, "", 0);
    }
    SvREFCNT_dec(cb);
    FREETMPS;
    LEAVE;
}

/* Invoke a callback with (undef) — EOF signal. Decrements cb refcount. */
static void invoke_eof_cb(SV *cb) {
    dSP;
    ENTER;
    SAVETMPS;
    PUSHMARK(SP);
    PUSHs(&PL_sv_undef);
    PUTBACK;
    call_sv(cb, G_DISCARD | G_EVAL);
    if (SvTRUE(ERRSV)) {
        warn("EV::MariaDB: exception in stream callback: %s", SvPV_nolen(ERRSV));
        sv_setpvn(ERRSV, "", 0);
    }
    SvREFCNT_dec(cb);
    FREETMPS;
    LEAVE;
}

/* Invoke a callback SV with args already on the stack. Decrements refcount. */
static void invoke_cb(SV *cb) {
    call_sv(cb, G_DISCARD | G_EVAL);
    if (SvTRUE(ERRSV)) {
        warn("EV::MariaDB: exception in callback: %s", SvPV_nolen(ERRSV));
        sv_setpvn(ERRSV, "", 0);
    }
    SvREFCNT_dec(cb);
}

/* Returns 1 if self was freed (caller must not touch self). */
static int deliver_result(ev_mariadb_t *self) {
    MYSQL_RES *res = self->op_result;
    SV *cb = pop_cb(self);

    if (cb == NULL) {
        if (res) {
            mysql_free_result(res);
            self->op_result = NULL;
        }
        return 0;
    }

    self->callback_depth++;

    {
        dSP;
        ENTER;
        SAVETMPS;
        PUSHMARK(SP);

        if (res == NULL && mysql_field_count(self->conn) > 0) {
            PUSHs(&PL_sv_undef);
            PUSHs(sv_2mortal(newSVpv(mysql_error(self->conn), 0)));
        }
        else if (res != NULL) {
            my_ulonglong nrows = mysql_num_rows(res);
            unsigned int ncols = mysql_num_fields(res);
            MYSQL_FIELD *fields = mysql_fetch_fields(res);
            AV *rows = newAV();
            AV *fnames;
            MYSQL_ROW row;
            unsigned long *lengths;

            if (nrows > 0) av_extend(rows, (SSize_t)(nrows < (my_ulonglong)SSize_t_MAX ? nrows - 1 : SSize_t_MAX));
            while ((row = mysql_fetch_row(res)) != NULL) {
                AV *r = newAV();
                unsigned int c;
                lengths = mysql_fetch_lengths(res);
                if (ncols > 0) av_extend(r, ncols - 1);
                for (c = 0; c < ncols; c++) {
                    if (row[c] == NULL) {
                        av_push(r, newSV(0));
                    } else {
                        SV *val = newSVpvn(row[c], lengths[c]);
                        if (self->utf8 && is_utf8_charset(fields[c].charsetnr))
                            SvUTF8_on(val);
                        av_push(r, val);
                    }
                }
                av_push(rows, newRV_noinc((SV*)r));
            }
            /* build field names before freeing result */
            fnames = build_field_names(self->conn, fields, ncols);
            mysql_free_result(res);
            self->op_result = NULL;
            PUSHs(sv_2mortal(newRV_noinc((SV*)rows)));
            PUSHs(&PL_sv_undef);
            PUSHs(sv_2mortal(newRV_noinc((SV*)fnames)));
        }
        else {
            my_ulonglong affected = mysql_affected_rows(self->conn);
            PUSHs(sv_2mortal(newSVuv((UV)affected)));
        }

        PUTBACK;
        invoke_cb(cb);
        FREETMPS;
        LEAVE;
    }

    self->callback_depth--;
    return check_destroyed(self);
}

/* Returns 1 if self was freed. */
static int deliver_error(ev_mariadb_t *self, const char *errmsg) {
    SV *cb = pop_cb(self);
    if (cb == NULL) return 0;

    self->callback_depth++;
    invoke_error_cb(cb, errmsg);
    self->callback_depth--;
    return check_destroyed(self);
}

/* Returns 1 if self was freed. Caller passes a non-mortal SV; ownership is
   transferred — deliver_value will sv_2mortal inside its SAVETMPS scope so
   FREETMPS correctly frees it (avoids mortal leak when sv_2mortal is called
   before SAVETMPS). */
static int deliver_value(ev_mariadb_t *self, SV *val) {
    SV *cb = pop_cb(self);
    if (cb == NULL) {
        SvREFCNT_dec(val);
        return 0;
    }

    self->callback_depth++;

    {
        dSP;
        ENTER;
        SAVETMPS;
        PUSHMARK(SP);
        PUSHs(sv_2mortal(val));
        PUTBACK;
        invoke_cb(cb);
        FREETMPS;
        LEAVE;
    }

    self->callback_depth--;
    return check_destroyed(self);
}

static void cleanup_connection(ev_mariadb_t *self) {
    int saved_fd = self->fd;
    int is_fork = IS_FORKED(self);

    stop_reading(self);
    stop_writing(self);
    stop_timer(self);
    self->fd = -1;
    self->state = STATE_IDLE;
    self->send_count = 0;
    self->draining = 0;

    if (self->op_data_ptr) {
        Safefree(self->op_data_ptr);
        self->op_data_ptr = NULL;
    }

    /* Discard any pending utility-op cache update — the op was cancelled. */
    FREE_STR(self->pending_user);
    FREE_STR(self->pending_password);
    FREE_STR(self->pending_database);
    FREE_STR(self->pending_charset);

    self->op_stmt_ctx = NULL;

    /* Invalidate all tracked stmt wrappers (don't free — user may hold handles) */
    {
        ev_mariadb_stmt_t *ctx = self->stmt_list;
        while (ctx) {
            free_stmt_bind_params(ctx);
            if (!is_fork && ctx->stmt) {
                if (ctx->stmt == self->op_stmt)
                    self->op_stmt = NULL;  /* avoid double-close below */
                mysql_stmt_close(ctx->stmt);
            }
            ctx->stmt = NULL;
            ctx->closed = 1;
            ctx = ctx->next;
        }
        /* keep stmt_list linked for close_stmt to unlink+free later */
    }

    if (is_fork) {
        self->op_result = NULL;
        self->op_stmt = NULL;
        self->conn = NULL;
    } else {
        if (self->op_result && saved_fd >= 0)
            shutdown(saved_fd, SHUT_RDWR);
        /* Close op_stmt if not tracked in stmt_list (e.g., in-flight prepare) */
        if (self->op_stmt) {
            mysql_stmt_close(self->op_stmt);
        }
        self->op_stmt = NULL;
        if (self->op_result) {
            mysql_free_result(self->op_result);
            self->op_result = NULL;
        }
        if (self->conn) {
            MYSQL *conn = self->conn;
            self->conn = NULL;
            mysql_close(conn);
        }
    }
}

static void cancel_pending(ev_mariadb_t *self, const char *errmsg) {
    ngx_queue_t local_send;
    ngx_queue_t local_cb;

    ngx_queue_init(&local_send);
    ngx_queue_init(&local_cb);

    if (!ngx_queue_empty(&self->send_queue)) {
        ngx_queue_add(&local_send, &self->send_queue);
        ngx_queue_init(&self->send_queue);
    }

    if (!ngx_queue_empty(&self->cb_queue)) {
        ngx_queue_add(&local_cb, &self->cb_queue);
        ngx_queue_init(&self->cb_queue);
    }

    self->send_count = 0;
    self->callback_depth++;

    /* cancel active stream if any */
    if (self->stream_cb) {
        SV *cb = self->stream_cb;
        self->stream_cb = NULL;
        self->pending_count--;
        invoke_error_cb(cb, errmsg);
    }

    /* cancel unsent queries */
    while (!ngx_queue_empty(&local_send)) {
        ngx_queue_t *q = ngx_queue_head(&local_send);
        ev_mariadb_send_t *send = ngx_queue_data(q, ev_mariadb_send_t, queue);
        SV *cb = send->cb;
        ngx_queue_remove(q);
        Safefree(send->sql);
        release_send(send);
        self->pending_count--;

        invoke_error_cb(cb, errmsg);
    }

    /* cancel sent-but-not-received */
    while (!ngx_queue_empty(&local_cb)) {
        ngx_queue_t *q = ngx_queue_head(&local_cb);
        ev_mariadb_cb_t *cbt = ngx_queue_data(q, ev_mariadb_cb_t, queue);
        SV *cb = cbt->cb;
        ngx_queue_remove(q);
        self->pending_count--;
        release_cbt(cbt);

        invoke_error_cb(cb, errmsg);
    }

    self->callback_depth--;
}

static void push_cb(ev_mariadb_t *self, SV *cb) {
    ev_mariadb_cb_t *cbt = alloc_cbt();
    cbt->cb = SvREFCNT_inc(cb);
    ngx_queue_insert_tail(&self->cb_queue, &cbt->queue);
    self->pending_count++;
}

/* Push cb to cb_queue, transferring ownership (no extra SvREFCNT_inc). */
static void push_cb_owned(ev_mariadb_t *self, SV *cb) {
    ev_mariadb_cb_t *cbt = alloc_cbt();
    cbt->cb = cb;
    ngx_queue_insert_tail(&self->cb_queue, &cbt->queue);
    /* pending_count was incremented by push_send; ownership transfers from send_queue to cb_queue */
}

static SV* handler_accessor(SV **slot, SV *handler, int has_arg) {
    if (has_arg) {
        if (NULL != *slot) {
            SvREFCNT_dec(*slot);
            *slot = NULL;
        }
        if (NULL != handler && SvOK(handler) &&
            SvROK(handler) && SvTYPE(SvRV(handler)) == SVt_PVCV) {
            *slot = SvREFCNT_inc(handler);
        }
    }

    return (NULL != *slot)
        ? SvREFCNT_inc(*slot)
        : &PL_sv_undef;
}

/* --- Multi-result drain --- */

static void drain_multi_result(ev_mariadb_t *self) {
    int status;

    self->draining = 1;
    status = mysql_next_result_start(&self->op_ret, self->conn);
    if (status != 0) {
        self->state = STATE_NEXT_RESULT;
        update_watchers(self, status);
        return;
    }
    on_next_result_done(self);
}

static void on_next_result_done(ev_mariadb_t *self) {
    int status;

    for (;;) {
        if (self->op_ret > 0) {
            /* error in secondary result set — report and stop draining */
            self->draining = 0;
            self->state = STATE_IDLE;
            self->callback_depth++;
            emit_error(self, mysql_error(self->conn));
            self->callback_depth--;
            if (check_destroyed(self)) return;
            if (self->state != STATE_IDLE) return;
            if (self->conn && mysql_more_results(self->conn)) {
                drain_multi_result(self);
                return;
            }
            pipeline_advance(self);
            return;
        }

        if (self->op_ret == -1) {
            /* no more results */
            break;
        }

        /* op_ret == 0: another result set is available */
        if (mysql_field_count(self->conn) > 0) {
            status = mysql_store_result_start(&self->op_result, self->conn);
            if (status != 0) {
                self->state = STATE_STORE_RESULT;
                update_watchers(self, status);
                return;
            }
            /* synchronous store — free and continue drain */
            if (self->op_result) {
                mysql_free_result(self->op_result);
                self->op_result = NULL;
            }
        }

        /* DML result or sync store done — check for more */
        if (!mysql_more_results(self->conn))
            break;

        status = mysql_next_result_start(&self->op_ret, self->conn);
        if (status != 0) {
            self->state = STATE_NEXT_RESULT;
            update_watchers(self, status);
            return;
        }
        /* synchronous completion — loop instead of recursing */
    }

    self->draining = 0;
    self->state = STATE_IDLE;
    pipeline_advance(self);
}

/* --- Pipeline: send_query + read_query_result state machine --- */

static void handle_send_failure(ev_mariadb_t *self, ev_mariadb_send_t *send) {
    char errbuf[512];
    SV *cb = send->cb;
    Safefree(send->sql);
    release_send(send);
    self->pending_count--;

    COPY_ERROR(errbuf, mysql_error(self->conn));

    self->state = STATE_IDLE;
    self->callback_depth++;
    invoke_error_cb(cb, errbuf);
    self->callback_depth--;
    if (check_destroyed(self)) return;
    if (self->state != STATE_IDLE) return;
    cancel_pending(self, "send failed");
    if (check_destroyed(self)) return;
    if (self->state != STATE_IDLE) return;
    drain_queues_silent(self);
    cleanup_connection(self);
}

static void on_send_done(ev_mariadb_t *self) {
    ngx_queue_t *q = ngx_queue_head(&self->send_queue);
    ev_mariadb_send_t *send = ngx_queue_data(q, ev_mariadb_send_t, queue);
    ngx_queue_remove(q);

    if (self->op_ret != 0) {
        handle_send_failure(self, send);
        return;
    }

    push_cb_owned(self, send->cb);
    Safefree(send->sql);
    release_send(send);
    self->send_count++;

    self->state = STATE_IDLE;
    pipeline_advance(self);
}

static void on_read_result_done(ev_mariadb_t *self) {
    int status;

    if (self->op_bool_ret != 0) {
        /* query returned an error */
        self->send_count--;
        self->state = STATE_IDLE;
        if (deliver_error(self, mysql_error(self->conn))) return;
        if (self->state != STATE_IDLE) return;
        /* drain any remaining result sets */
        if (self->conn && mysql_more_results(self->conn)) {
            drain_multi_result(self);
            return;
        }
        pipeline_advance(self);
        return;
    }

    if (mysql_field_count(self->conn) > 0) {
        /* has result set — store it */
        status = mysql_store_result_start(&self->op_result, self->conn);
        if (status != 0) {
            self->state = STATE_STORE_RESULT;
            update_watchers(self, status);
            return;
        }
        /* synchronous store completion — fall through below */
    }

    /* DML or synchronous store_result completion */
    {
        self->send_count--;
        self->state = STATE_IDLE;
        if (deliver_result(self)) return;
        if (self->state != STATE_IDLE) return;
        /* drain any extra result sets from multi-result queries */
        if (self->conn && mysql_more_results(self->conn)) {
            drain_multi_result(self);
            return;
        }
        pipeline_advance(self);
    }
}

/* Called when store_result_cont completes (pipeline text query path) */
static void on_store_result_done(ev_mariadb_t *self) {
    if (self->draining) {
        MYSQL_RES *res = self->op_result;
        self->op_result = NULL;
        /* draining multi-result: free and continue */
        if (res) mysql_free_result(res);
        if (self->conn && mysql_more_results(self->conn)) {
            int status = mysql_next_result_start(&self->op_ret, self->conn);
            if (status != 0) {
                self->state = STATE_NEXT_RESULT;
                update_watchers(self, status);
                return;
            }
            on_next_result_done(self);
            return;
        }
        self->draining = 0;
        self->state = STATE_IDLE;
        pipeline_advance(self);
        return;
    }

    self->send_count--;
    self->state = STATE_IDLE;

    if (deliver_result(self)) return;
    if (self->state != STATE_IDLE) return;
    /* drain any extra result sets */
    if (self->conn && mysql_more_results(self->conn)) {
        drain_multi_result(self);
        return;
    }
    pipeline_advance(self);
}

/*
 * Pipeline orchestrator. Called when state == IDLE.
 * Phase 1: send all queued queries via mysql_send_query.
 * Phase 2: read next result via mysql_read_query_result.
 */
static void pipeline_advance(ev_mariadb_t *self) {
    int status;

    /* Ensure clean watcher state — previous operations may have left
     * watchers active after completing synchronously within their
     * done handlers. Without this, subsequent operations that need
     * the same watcher direction would skip ev_io_start. */
    stop_reading(self);
    stop_writing(self);
    stop_timer(self);

send_phase:
    /* Phase 1: send up to MAX_PIPELINE_DEPTH queries */
    while (!ngx_queue_empty(&self->send_queue) &&
           self->send_count < MAX_PIPELINE_DEPTH) {
        ngx_queue_t *q = ngx_queue_head(&self->send_queue);
        ev_mariadb_send_t *send = ngx_queue_data(q, ev_mariadb_send_t, queue);

        status = mysql_send_query_start(&self->op_ret, self->conn,
            send->sql, send->sql_len);

        if (status != 0) {
            /* need async IO to finish sending */
            self->state = STATE_SEND;
            update_watchers(self, status);
            return;
        }

        /* synchronous send completion */
        ngx_queue_remove(q);

        if (self->op_ret != 0) {
            handle_send_failure(self, send);
            return;
        }

        push_cb_owned(self, send->cb);
        Safefree(send->sql);
        release_send(send);
        self->send_count++;
    }

    /* Phase 2: read next result */
    while (self->send_count > 0) {
        status = mysql_read_query_result_start(&self->op_bool_ret, self->conn);

        if (status != 0) {
            self->state = STATE_READ_RESULT;
            update_watchers(self, status);
            return;
        }

        /* synchronous read completion */
        if (self->op_bool_ret != 0) {
            /* query error */
            self->send_count--;
            if (deliver_error(self, mysql_error(self->conn))) return;
            /* callback may have started a new operation */
            if (self->state != STATE_IDLE) return;
            if (self->conn && mysql_more_results(self->conn)) {
                drain_multi_result(self);
                return;
            }
            if (!ngx_queue_empty(&self->send_queue)) goto send_phase;
            continue;
        }

        if (mysql_field_count(self->conn) > 0) {
            /* has result set */
            status = mysql_store_result_start(&self->op_result, self->conn);
            if (status != 0) {
                self->state = STATE_STORE_RESULT;
                update_watchers(self, status);
                return;
            }
        }

        {
            self->send_count--;
            if (deliver_result(self)) return;
            /* callback may have started a new operation */
            if (self->state != STATE_IDLE) return;
            if (self->conn && mysql_more_results(self->conn)) {
                drain_multi_result(self);
                return;
            }
            /* loop back to check send_queue (callback may have queued more) */
            if (!ngx_queue_empty(&self->send_queue)) goto send_phase;
        }
    }

    self->state = STATE_IDLE;
}

/* --- Connection --- */

static void on_connect_done(ev_mariadb_t *self) {
    self->state = STATE_IDLE;

    if (self->op_conn_ret == NULL) {
        char errbuf[512];
        COPY_ERROR(errbuf, mysql_error(self->conn));
        self->callback_depth++;
        emit_error(self, errbuf);
        self->callback_depth--;
        if (check_destroyed(self)) return;
        /* on_error handler may have called reset(), starting a new connection;
           if state is no longer IDLE, don't touch the new connection */
        if (self->state != STATE_IDLE) return;
        cancel_pending(self, errbuf);
        if (check_destroyed(self)) return;
        if (self->state != STATE_IDLE) return;
        drain_queues_silent(self);
        cleanup_connection(self);
        return;
    }

    /* connected — reinit watchers for normal IO */
    stop_reading(self);
    stop_writing(self);
    stop_timer(self);

    self->fd = mysql_get_socket(self->conn);

    if (self->fd < 0) {
        self->callback_depth++;
        emit_error(self, "mysql_get_socket returned invalid fd");
        self->callback_depth--;
        if (check_destroyed(self)) return;
        if (self->state != STATE_IDLE) return;
        cancel_pending(self, "invalid fd");
        if (check_destroyed(self)) return;
        if (self->state != STATE_IDLE) return;
        drain_queues_silent(self);
        cleanup_connection(self);
        return;
    }

    init_io_watchers(self);

    if (NULL != self->on_connect) {
        SV *cb = sv_2mortal(SvREFCNT_inc_simple_NN(self->on_connect));
        self->callback_depth++;

        {
            dSP;
            ENTER;
            SAVETMPS;
            PUSHMARK(SP);
            PUTBACK;

            call_sv(cb, G_DISCARD | G_EVAL);
            if (SvTRUE(ERRSV)) {
                warn("EV::MariaDB: exception in connect handler: %s", SvPV_nolen(ERRSV));
                sv_setpvn(ERRSV, "", 0);
            }

            FREETMPS;
            LEAVE;
        }

        self->callback_depth--;
        if (check_destroyed(self)) return;
    }

    /* start pipeline if queries were queued during connection */
    maybe_pipeline(self);
}

/* --- Prepared statements --- */

static void on_stmt_prepare_done(ev_mariadb_t *self) {
    MYSQL_STMT *stmt = self->op_stmt;
    self->op_stmt = NULL;
    self->state = STATE_IDLE;

    if (self->op_ret != 0) {
        char errbuf[512];
        COPY_ERROR(errbuf, mysql_stmt_error(stmt));
        mysql_stmt_close(stmt);
        if (deliver_error(self, errbuf)) return;
        maybe_pipeline(self);
        return;
    }

    {
        ev_mariadb_stmt_t *ctx;
        Newxz(ctx, 1, ev_mariadb_stmt_t);
        ctx->stmt = stmt;
        ctx->next = self->stmt_list;
        self->stmt_list = ctx;
        if (deliver_value(self, newSViv(PTR2IV(ctx)))) return;
    }
    maybe_pipeline(self);
}

static void on_stmt_execute_done(ev_mariadb_t *self) {
    MYSQL_STMT *stmt = self->op_stmt;
    self->op_stmt = NULL;
    self->state = STATE_IDLE;

    if (self->op_ret != 0) {
        if (deliver_error(self, mysql_stmt_error(stmt))) return;
        maybe_pipeline(self);
        return;
    }

    {
        MYSQL_RES *meta;
        SV *cb;
        meta = mysql_stmt_result_metadata(stmt);
        self->op_result = meta;
        cb = pop_cb(self);
        if (cb == NULL) {
            if (self->op_result) {
                mysql_stmt_free_result(stmt);
                mysql_free_result(self->op_result);
                self->op_result = NULL;
            }
            maybe_pipeline(self);
            return;
        }

        self->callback_depth++;
        {
            dSP;
            ENTER;
            SAVETMPS;
            PUSHMARK(SP);

            if (self->op_result == NULL) {
                my_ulonglong affected = mysql_stmt_affected_rows(stmt);
                PUSHs(sv_2mortal(newSVuv((UV)affected)));
            }
            else {
                unsigned int ncols = mysql_num_fields(self->op_result);
                MYSQL_FIELD *fields = mysql_fetch_fields(self->op_result);
                MYSQL_BIND *bind;
                unsigned long *lengths;
                my_bool *is_null;
                char **buffers;
                unsigned int c;
                AV *rows = newAV();
                int fetch_ret;

                Newxz(bind, ncols, MYSQL_BIND);
                SAVEFREEPV(bind);
                Newx(lengths, ncols, unsigned long);
                SAVEFREEPV(lengths);
                Newx(is_null, ncols, my_bool);
                SAVEFREEPV(is_null);
                Newx(buffers, ncols, char *);
                SAVEFREEPV(buffers);

                for (c = 0; c < ncols; c++) {
                    unsigned long buflen = fields[c].max_length;
                    if (buflen < 256) buflen = 256;
                    Newx(buffers[c], buflen, char);
                    SAVEFREEPV(buffers[c]);
                    bind[c].buffer_type = MYSQL_TYPE_STRING;
                    bind[c].buffer = buffers[c];
                    bind[c].buffer_length = buflen;
                    bind[c].length = &lengths[c];
                    bind[c].is_null = &is_null[c];
                }

                if (mysql_stmt_bind_result(stmt, bind)) {
                    mysql_stmt_free_result(stmt);
                    mysql_free_result(self->op_result);
                    self->op_result = NULL;
                    SvREFCNT_dec((SV*)rows);

                    PUSHs(&PL_sv_undef);
                    PUSHs(sv_2mortal(newSVpv(mysql_stmt_error(stmt), 0)));
                    goto invoke;
                }

                while ((fetch_ret = mysql_stmt_fetch(stmt)) == 0) {
                    AV *row = newAV();
                    if (ncols > 0) av_extend(row, ncols - 1);
                    for (c = 0; c < ncols; c++) {
                        if (is_null[c]) {
                            av_push(row, newSV(0));
                        } else {
                            SV *val = newSVpvn(buffers[c], lengths[c]);
                            if (self->utf8 && is_utf8_charset(fields[c].charsetnr))
                                SvUTF8_on(val);
                            av_push(row, val);
                        }
                    }
                    av_push(rows, newRV_noinc((SV*)row));
                }

                if (fetch_ret != 0 && fetch_ret != MYSQL_NO_DATA) {
                    mysql_free_result(self->op_result);
                    self->op_result = NULL;
                    mysql_stmt_free_result(stmt);
                    SvREFCNT_dec((SV*)rows);
                    PUSHs(&PL_sv_undef);
                    if (fetch_ret == MYSQL_DATA_TRUNCATED)
                        PUSHs(sv_2mortal(newSVpvs("data truncated")));
                    else
                        PUSHs(sv_2mortal(newSVpv(mysql_stmt_error(stmt), 0)));
                } else {
                    AV *fnames = build_field_names(self->conn, fields, ncols);
                    mysql_free_result(self->op_result);
                    self->op_result = NULL;
                    mysql_stmt_free_result(stmt);
                    PUSHs(sv_2mortal(newRV_noinc((SV*)rows)));
                    PUSHs(&PL_sv_undef);
                    PUSHs(sv_2mortal(newRV_noinc((SV*)fnames)));
                }
            }

        invoke:
            PUTBACK;
            invoke_cb(cb);
            FREETMPS;
            LEAVE;
        }
        self->callback_depth--;
        if (check_destroyed(self)) return;
        maybe_pipeline(self);
    }
}

static void on_stmt_store_done(ev_mariadb_t *self) {
    if (self->op_ret != 0) {
        MYSQL_STMT *stmt = self->op_stmt;
        self->op_stmt = NULL;
        self->state = STATE_IDLE;
        if (deliver_error(self, mysql_stmt_error(stmt))) return;
        maybe_pipeline(self);
        return;
    }
    on_stmt_execute_done(self);
}

static void unlink_stmt(ev_mariadb_t *self, ev_mariadb_stmt_t *ctx) {
    ev_mariadb_stmt_t **pp = &self->stmt_list;
    while (*pp) {
        if (*pp == ctx) { *pp = ctx->next; return; }
        pp = &(*pp)->next;
    }
}

static void on_stmt_close_done(ev_mariadb_t *self) {
    self->op_stmt = NULL;  /* stmt freed by mysql_stmt_close */
    if (self->op_stmt_ctx) {
        unlink_stmt(self, self->op_stmt_ctx);
        Safefree(self->op_stmt_ctx);
        self->op_stmt_ctx = NULL;
    }
    self->state = STATE_IDLE;
    if (self->op_bool_ret != 0) {
        if (deliver_error(self, mysql_error(self->conn))) return;
    } else {
        if (deliver_value(self, newSViv(1))) return;
    }
    maybe_pipeline(self);
}

static void on_stmt_reset_done(ev_mariadb_t *self) {
    MYSQL_STMT *stmt = self->op_stmt;
    self->op_stmt = NULL;
    self->state = STATE_IDLE;
    if (self->op_bool_ret != 0) {
        if (deliver_error(self, mysql_stmt_error(stmt))) return;
    } else {
        if (deliver_value(self, newSViv(1))) return;
    }
    maybe_pipeline(self);
}

/* --- Async utility operation done handler --- */

/* mysql_reset_connection / mysql_change_user drop all server-side prepared
   statements; invalidate every tracked wrapper so user code can't keep using
   them. Closes the local libmariadb stmt state and marks closed=1, mirroring
   cleanup_connection's pattern but without touching the live connection. */
static void invalidate_all_stmts(ev_mariadb_t *self) {
    ev_mariadb_stmt_t *ctx = self->stmt_list;
    while (ctx) {
        free_stmt_bind_params(ctx);
        if (ctx->stmt) {
            mysql_stmt_close(ctx->stmt);
            ctx->stmt = NULL;
        }
        ctx->closed = 1;
        ctx = ctx->next;
    }
    self->op_stmt = NULL;
    self->op_stmt_ctx = NULL;
}

/* Commit (success) or discard (failure) any pending utility-op cache update. */
static void resolve_pending_cache(ev_mariadb_t *self, int failed) {
    if (failed) {
        FREE_STR(self->pending_user);
        FREE_STR(self->pending_password);
        FREE_STR(self->pending_database);
        FREE_STR(self->pending_charset);
        return;
    }
    if (self->pending_user) {
        Safefree(self->user);
        self->user = self->pending_user;
        self->pending_user = NULL;
    }
    if (self->pending_password) {
        Safefree(self->password);
        self->password = self->pending_password;
        self->pending_password = NULL;
    }
    if (self->pending_database) {
        Safefree(self->database);
        self->database = self->pending_database;
        self->pending_database = NULL;
    }
    if (self->pending_charset) {
        Safefree(self->charset);
        self->charset = self->pending_charset;
        self->pending_charset = NULL;
    }
}

static void on_utility_done(ev_mariadb_t *self, int failed) {
    enum ev_mariadb_state prev = self->state;
    self->state = STATE_IDLE;
    resolve_pending_cache(self, failed);
    if (!failed && (prev == STATE_RESET_CONNECTION || prev == STATE_CHANGE_USER))
        invalidate_all_stmts(self);
    if (failed) {
        if (deliver_error(self, mysql_error(self->conn))) return;
    } else {
        if (deliver_value(self, newSViv(1))) return;
    }
    maybe_pipeline(self);
}

/* Trailing dispatch shared by every utility op start/cont site:
   completed → on_utility_done; otherwise → arm watchers. */
static void dispatch_utility(ev_mariadb_t *self, int status, int failed) {
    if (status == 0) on_utility_done(self, failed);
    else update_watchers(self, status);
}

static void free_stmt_bind_params(ev_mariadb_stmt_t *ctx) {
    int i;
    if (!ctx->bind_params) return;
    for (i = 0; i < ctx->bind_param_count; i++) {
        Safefree(ctx->bind_params[i].buffer);
    }
    Safefree(ctx->bind_params);
    ctx->bind_params = NULL;
    ctx->bind_param_count = 0;
}

static void transition_to_stmt_store(ev_mariadb_t *self) {
    int status;
    self->op_stmt_ctx = NULL;  /* bind_params live in wrapper, not freed here */
    if (self->op_ret != 0) {
        on_stmt_execute_done(self);
    } else {
        self->state = STATE_STMT_STORE;
        status = mysql_stmt_store_result_start(&self->op_ret, self->op_stmt);
        if (status == 0) {
            on_stmt_store_done(self);
        } else {
            update_watchers(self, status);
        }
    }
}

/* --- Streaming query (mysql_use_result + fetch_row) --- */

static void stream_error(ev_mariadb_t *self) {
    SV *cb = self->stream_cb;
    char errbuf[512];
    COPY_ERROR(errbuf, mysql_error(self->conn));
    self->stream_cb = NULL;
    self->state = STATE_IDLE;
    self->pending_count--;
    self->callback_depth++;
    invoke_error_cb(cb, errbuf);
    self->callback_depth--;
}

static void on_real_query_done(ev_mariadb_t *self) {
    int status;

    if (self->op_ret != 0) {
        stream_error(self);
        if (check_destroyed(self)) return;
        maybe_pipeline(self);
        return;
    }

    self->op_result = mysql_use_result(self->conn);
    if (!self->op_result) {
        const char *err = mysql_error(self->conn);
        if (err && err[0]) {
            stream_error(self);
            if (check_destroyed(self)) return;
        } else {
            /* DML or no-result query: deliver EOF to stream callback */
            SV *cb = self->stream_cb;
            self->stream_cb = NULL;
            self->state = STATE_IDLE;
            self->pending_count--;
            self->callback_depth++;
            invoke_eof_cb(cb);
            self->callback_depth--;
            if (check_destroyed(self)) return;
        }
        maybe_pipeline(self);
        return;
    }

    status = mysql_fetch_row_start(&self->op_row, self->op_result);
    if (status == 0) {
        on_stream_fetch_done(self);
    } else {
        self->state = STATE_STREAM_FETCH;
        update_watchers(self, status);
    }
}

static void on_stream_fetch_done(ev_mariadb_t *self) {
    MYSQL_FIELD *fields = (self->utf8 && self->op_result)
        ? mysql_fetch_fields(self->op_result) : NULL;

    for (;;) {
        MYSQL_ROW row = self->op_row;

        if (row == NULL) break;

        /* deliver row to callback */
        {
            unsigned int ncols = mysql_num_fields(self->op_result);
            unsigned long *lengths = mysql_fetch_lengths(self->op_result);
            AV *r = newAV();
            unsigned int c;

            if (ncols > 0) av_extend(r, ncols - 1);
            for (c = 0; c < ncols; c++) {
                if (row[c] == NULL) {
                    av_push(r, newSV(0));
                } else {
                    SV *val = newSVpvn(row[c], lengths[c]);
                    if (fields && is_utf8_charset(fields[c].charsetnr))
                        SvUTF8_on(val);
                    av_push(r, val);
                }
            }

            /* Detach stream_cb before invoking: prevents cancel_pending
               from double-firing the callback if finish() is called inside */
            {
                SV *saved_cb = self->stream_cb;
                self->stream_cb = NULL;
                self->pending_count--;

                self->callback_depth++;
                {
                    SV *cb = sv_2mortal(SvREFCNT_inc_simple_NN(saved_cb));
                    dSP;
                    ENTER;
                    SAVETMPS;
                    PUSHMARK(SP);
                    PUSHs(sv_2mortal(newRV_noinc((SV*)r)));
                    PUTBACK;
                    call_sv(cb, G_DISCARD | G_EVAL);
                    if (SvTRUE(ERRSV)) {
                        warn("EV::MariaDB: exception in stream callback: %s", SvPV_nolen(ERRSV));
                        sv_setpvn(ERRSV, "", 0);
                    }
                    FREETMPS;
                    LEAVE;
                }
                self->callback_depth--;
                if (check_destroyed(self)) {
                    SvREFCNT_dec(saved_cb);
                    return;
                }

                /* Was stream cancelled (finish/reset from callback)? */
                if (self->stream_cb != NULL) {
                    /* Defence: unreachable via the public API — query_stream's
                       CHECK_READY rejects the only way stream_cb could be
                       reset while we're inside on_stream_fetch_done. Drop the
                       saved cb so the (hypothetical) new stream owns the slot. */
                    SvREFCNT_dec(saved_cb);
                } else if (self->conn == NULL || self->op_result == NULL) {
                    /* connection torn down — stream is over */
                    SvREFCNT_dec(saved_cb);
                    return;
                } else {
                    /* normal: restore for next row */
                    self->stream_cb = saved_cb;
                    self->pending_count++;
                }
            }
        }

        /* fetch next row */
        {
            int status = mysql_fetch_row_start(&self->op_row, self->op_result);
            if (status != 0) {
                self->state = STATE_STREAM_FETCH;
                update_watchers(self, status);
                return;
            }
        }
    }

    /* row == NULL: EOF or error */
    {
        SV *cb = self->stream_cb;
        char errbuf[512];
        const char *err = mysql_error(self->conn);
        int has_error = (err && err[0]);
        if (has_error) COPY_ERROR(errbuf, err);

        self->stream_cb = NULL;
        mysql_free_result(self->op_result);
        self->op_result = NULL;
        self->state = STATE_IDLE;
        self->pending_count--;

        self->callback_depth++;
        if (has_error) {
            invoke_error_cb(cb, errbuf);
        } else {
            invoke_eof_cb(cb);
        }
        self->callback_depth--;
        if (check_destroyed(self)) return;
        /* drain any secondary result sets from multi_statements */
        if (self->conn && mysql_more_results(self->conn)) {
            drain_multi_result(self);
            return;
        }
        maybe_pipeline(self);
    }
}

/* --- Async close --- */

static void on_close_done(ev_mariadb_t *self) {
    SV *cb;

    /* conn was freed by mysql_close */
    self->conn = NULL;
    self->fd = -1;
    stop_reading(self);
    stop_writing(self);
    stop_timer(self);
    self->state = STATE_IDLE;

    /* pop our callback before cancel_pending fires errors for remaining items */
    cb = pop_cb(self);

    /* cancel any items queued from within a callback before close_async */
    cancel_pending(self, "connection closed");
    if (check_destroyed(self)) {
        SvREFCNT_dec(cb);
        return;
    }

    if (cb) {
        SV *val = newSViv(1);
        self->callback_depth++;
        {
            dSP;
            ENTER;
            SAVETMPS;
            PUSHMARK(SP);
            PUSHs(sv_2mortal(val));
            PUTBACK;
            invoke_cb(cb);
            FREETMPS;
            LEAVE;
        }
        self->callback_depth--;
        if (check_destroyed(self)) return;
    }
    /* no maybe_pipeline: conn is NULL, nothing can be sent */
}

/* --- Send long data done --- */

static void on_send_long_data_done(ev_mariadb_t *self) {
    MYSQL_STMT *stmt = self->op_stmt;

    if (self->op_data_ptr) {
        Safefree(self->op_data_ptr);
        self->op_data_ptr = NULL;
    }
    self->op_stmt = NULL;
    self->state = STATE_IDLE;

    if (self->op_bool_ret != 0) {
        if (deliver_error(self, mysql_stmt_error(stmt))) return;
    } else {
        if (deliver_value(self, newSViv(1))) return;
    }
    maybe_pipeline(self);
}

/* --- Main continuation dispatcher --- */

static void continue_operation(ev_mariadb_t *self, int events) {
    int status;

    switch (self->state) {
    case STATE_CONNECTING:
        status = mysql_real_connect_cont(&self->op_conn_ret, self->conn, events);
        if (status == 0) {
            on_connect_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_SEND:
        status = mysql_send_query_cont(&self->op_ret, self->conn, events);
        if (status == 0) {
            on_send_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_READ_RESULT:
        status = mysql_read_query_result_cont(&self->op_bool_ret, self->conn, events);
        if (status == 0) {
            on_read_result_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STORE_RESULT:
        status = mysql_store_result_cont(&self->op_result, self->conn, events);
        if (status == 0) {
            on_store_result_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_NEXT_RESULT:
        status = mysql_next_result_cont(&self->op_ret, self->conn, events);
        if (status == 0) {
            on_next_result_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STMT_PREPARE:
        status = mysql_stmt_prepare_cont(&self->op_ret, self->op_stmt, events);
        if (status == 0) {
            on_stmt_prepare_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STMT_EXECUTE:
        status = mysql_stmt_execute_cont(&self->op_ret, self->op_stmt, events);
        if (status == 0) {
            transition_to_stmt_store(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STMT_STORE:
        status = mysql_stmt_store_result_cont(&self->op_ret, self->op_stmt, events);
        if (status == 0) {
            on_stmt_store_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STMT_CLOSE:
        status = mysql_stmt_close_cont(&self->op_bool_ret, self->op_stmt, events);
        if (status == 0) {
            on_stmt_close_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STMT_RESET:
        status = mysql_stmt_reset_cont(&self->op_bool_ret, self->op_stmt, events);
        if (status == 0) {
            on_stmt_reset_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_PING:
        status = mysql_ping_cont(&self->op_ret, self->conn, events);
        dispatch_utility(self, status, self->op_ret != 0);
        break;

    case STATE_CHANGE_USER:
        status = mysql_change_user_cont(&self->op_bool_ret, self->conn, events);
        dispatch_utility(self, status, self->op_bool_ret != 0);
        break;

    case STATE_SELECT_DB:
        status = mysql_select_db_cont(&self->op_ret, self->conn, events);
        dispatch_utility(self, status, self->op_ret != 0);
        break;

    case STATE_RESET_CONNECTION:
        status = mysql_reset_connection_cont(&self->op_ret, self->conn, events);
        dispatch_utility(self, status, self->op_ret != 0);
        break;

    case STATE_SET_CHARSET:
        status = mysql_set_character_set_cont(&self->op_ret, self->conn, events);
        dispatch_utility(self, status, self->op_ret != 0);
        break;

    case STATE_COMMIT:
        status = mysql_commit_cont(&self->op_bool_ret, self->conn, events);
        dispatch_utility(self, status, self->op_bool_ret != 0);
        break;

    case STATE_ROLLBACK:
        status = mysql_rollback_cont(&self->op_bool_ret, self->conn, events);
        dispatch_utility(self, status, self->op_bool_ret != 0);
        break;

    case STATE_AUTOCOMMIT:
        status = mysql_autocommit_cont(&self->op_bool_ret, self->conn, events);
        dispatch_utility(self, status, self->op_bool_ret != 0);
        break;

    case STATE_STMT_SEND_LONG_DATA:
        status = mysql_stmt_send_long_data_cont(&self->op_bool_ret, self->op_stmt, events);
        if (status == 0) {
            on_send_long_data_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_REAL_QUERY:
        status = mysql_real_query_cont(&self->op_ret, self->conn, events);
        if (status == 0) {
            on_real_query_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_STREAM_FETCH:
        status = mysql_fetch_row_cont(&self->op_row, self->op_result, events);
        if (status == 0) {
            on_stream_fetch_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    case STATE_CLOSE:
        status = mysql_close_cont(self->conn, events);
        if (status == 0) {
            on_close_done(self);
        } else {
            update_watchers(self, status);
        }
        break;

    default:
        warn("EV::MariaDB: unexpected state %d in continue_operation", self->state);
        break;
    }
}

static void io_cb(EV_P_ ev_io *w, int revents) {
    ev_mariadb_t *self = (ev_mariadb_t *)w->data;
    int events = 0;
    (void)loop;

    if (self->magic != EV_MARIADB_MAGIC) return;
    if (NULL == self->conn) return;

    if (revents & EV_READ)  events |= MYSQL_WAIT_READ | MYSQL_WAIT_EXCEPT;
    if (revents & EV_WRITE) events |= MYSQL_WAIT_WRITE;

    continue_operation(self, events);
}

static void timer_cb(EV_P_ ev_timer *w, int revents) {
    ev_mariadb_t *self = (ev_mariadb_t *)w->data;
    (void)loop;
    (void)revents;

    if (self->magic != EV_MARIADB_MAGIC) return;
    if (NULL == self->conn) return;

    self->timing = 0;
    continue_operation(self, MYSQL_WAIT_TIMEOUT);
}

static char* safe_strdup(const char *s) {
    char *d;
    size_t len;
    if (!s) return NULL;
    len = strlen(s);
    Newx(d, len + 1, char);
    Copy(s, d, len + 1, char);
    return d;
}

/* newSVpv on a non-empty string, &PL_sv_undef otherwise. Used by accessors
   that wrap a libmariadb getter returning a NUL-terminated string or NULL. */
static SV* str_or_undef(const char *s) {
    return (s && s[0]) ? newSVpv(s, 0) : &PL_sv_undef;
}

/* Free *slot then replace it with a fresh copy of src (or NULL). */
static void replace_str(char **slot, const char *src) {
    Safefree(*slot);
    *slot = safe_strdup(src);
}

static void free_connect_strings(ev_mariadb_t *self) {
    FREE_STR(self->host);
    FREE_STR(self->user);
    FREE_STR(self->password);
    FREE_STR(self->database);
    FREE_STR(self->unix_socket);
}

static void free_option_strings(ev_mariadb_t *self) {
    FREE_STR(self->charset);
    FREE_STR(self->init_command);
    FREE_STR(self->ssl_key);
    FREE_STR(self->ssl_cert);
    FREE_STR(self->ssl_ca);
    FREE_STR(self->ssl_capath);
    FREE_STR(self->ssl_cipher);
}

static void apply_options(ev_mariadb_t *self) {
    MYSQL *conn = self->conn;
    unsigned long flags = self->client_flags;

    if (self->connect_timeout > 0)
        mysql_options(conn, MYSQL_OPT_CONNECT_TIMEOUT, &self->connect_timeout);
    if (self->read_timeout > 0)
        mysql_options(conn, MYSQL_OPT_READ_TIMEOUT, &self->read_timeout);
    if (self->write_timeout > 0)
        mysql_options(conn, MYSQL_OPT_WRITE_TIMEOUT, &self->write_timeout);
    if (self->compress)
        mysql_options(conn, MYSQL_OPT_COMPRESS, NULL);
    if (self->charset)
        mysql_options(conn, MYSQL_SET_CHARSET_NAME, self->charset);
    if (self->init_command)
        mysql_options(conn, MYSQL_INIT_COMMAND, self->init_command);
    if (self->ssl_ca || self->ssl_capath || self->ssl_cert || self->ssl_key || self->ssl_cipher)
        mysql_ssl_set(conn, self->ssl_key, self->ssl_cert, self->ssl_ca, self->ssl_capath, self->ssl_cipher);
    if (self->ssl_verify_server_cert) {
        my_bool val = 1;
        mysql_options(conn, MYSQL_OPT_SSL_VERIFY_SERVER_CERT, &val);
    }
    if (self->multi_statements)
        flags |= CLIENT_MULTI_STATEMENTS | CLIENT_MULTI_RESULTS;
    if (self->found_rows)
        flags |= CLIENT_FOUND_ROWS;

    self->client_flags = flags;
}

static int is_utf8_charset(unsigned int charsetnr) {
    /* utf8mb3 collations: 33, 83, 192-211 */
    /* utf8mb4 collations: 45, 46, 224-247, 255, 256-309 */
    /* MariaDB 10.10+ uca1400: utf8mb3 2048-2303, utf8mb4 2304-2559 */
    /* (2560+ are ucs2/utf16/utf32 uca1400 — NOT UTF-8) */
    return charsetnr == 33 || charsetnr == 83
        || (charsetnr >= 192 && charsetnr <= 211)
        || charsetnr == 45 || charsetnr == 46
        || (charsetnr >= 224 && charsetnr <= 247)
        || charsetnr == 255
        || (charsetnr >= 256 && charsetnr <= 309)
        || (charsetnr >= 2048 && charsetnr <= 2559);
}

static int conn_charset_is_utf8(MYSQL *conn) {
    const char *cs;
    if (!conn) return 0;
    cs = mysql_character_set_name(conn);
    return (cs && strncmp(cs, "utf8", 4) == 0);
}

static AV* build_field_names(MYSQL *conn, MYSQL_FIELD *fields, unsigned int ncols) {
    AV *fnames = newAV();
    unsigned int i;
    int utf8_names = conn_charset_is_utf8(conn);
    if (ncols > 0) av_extend(fnames, ncols - 1);
    for (i = 0; i < ncols; i++) {
        SV *name = newSVpvn(fields[i].name, fields[i].name_length);
        if (utf8_names)
            SvUTF8_on(name);
        av_push(fnames, name);
    }
    return fnames;
}

static void setup_bind_params(ev_mariadb_stmt_t *ctx, AV *params) {
    int nparams = (int)(av_len(params) + 1);
    MYSQL_BIND *bp;
    int i;

    {
        unsigned long expected = mysql_stmt_param_count(ctx->stmt);
        if ((unsigned long)nparams != expected)
            croak("parameter count mismatch: got %d, expected %lu", nparams, expected);
    }

    if (nparams == 0) return;

    free_stmt_bind_params(ctx);

    Newxz(bp, nparams, MYSQL_BIND);
    ctx->bind_params = bp;
    ctx->bind_param_count = nparams;

    for (i = 0; i < nparams; i++) {
        SV **svp = av_fetch(params, i, 0);
        if (svp && SvOK(*svp)) {
            if (SvIOK(*svp)) {
                long long *val;
                Newx(val, 1, long long);
                if (SvIsUV(*svp)) {
                    *val = (long long)(unsigned long long)SvUV(*svp);
                    bp[i].is_unsigned = 1;
                } else {
                    *val = (long long)SvIV(*svp);
                    bp[i].is_unsigned = 0;
                }
                bp[i].buffer_type = MYSQL_TYPE_LONGLONG;
                bp[i].buffer = (void *)val;
            } else if (SvNOK(*svp)) {
                double *val;
                Newx(val, 1, double);
                *val = (double)SvNV(*svp);
                bp[i].buffer_type = MYSQL_TYPE_DOUBLE;
                bp[i].buffer = (void *)val;
            } else {
                STRLEN len;
                const char *s = SvPV(*svp, len);
                char *copy;
                Newx(copy, len + 1, char);
                Copy(s, copy, len, char);
                copy[len] = '\0';
                bp[i].buffer_type = MYSQL_TYPE_STRING;
                bp[i].buffer = (void *)copy;
                bp[i].buffer_length = (unsigned long)len;
                bp[i].length = &bp[i].buffer_length;
            }
        } else {
            bp[i].buffer_type = MYSQL_TYPE_NULL;
        }
    }

    if (mysql_stmt_bind_param(ctx->stmt, bp)) {
        char errbuf[512];
        COPY_ERROR(errbuf, mysql_stmt_error(ctx->stmt));
        free_stmt_bind_params(ctx);
        croak("%s", errbuf);
    }
}

static void start_connect(ev_mariadb_t *self) {
    int status;

    self->conn = mysql_init(NULL);
    if (NULL == self->conn) {
        croak("mysql_init failed");
    }

    self->connect_pid = getpid();
    mysql_options(self->conn, MYSQL_OPT_NONBLOCK, 0);
    apply_options(self);

    self->state = STATE_CONNECTING;
    status = mysql_real_connect_start(&self->op_conn_ret, self->conn,
        self->host, self->user, self->password, self->database,
        self->port, self->unix_socket, self->client_flags);

    if (status == 0) {
        on_connect_done(self);
    } else {
        self->fd = mysql_get_socket(self->conn);
        if (self->fd < 0) {
            mysql_close(self->conn);
            self->conn = NULL;
            self->state = STATE_IDLE;
            croak("mysql_get_socket returned invalid fd");
        }

        init_io_watchers(self);

        update_watchers(self, status);
    }
}

MODULE = EV::MariaDB  PACKAGE = EV::MariaDB

BOOT:
{
    I_EV_API("EV::MariaDB");
    call_atexit(drain_freelists, NULL);
}

EV::MariaDB
_new(char *class, EV::Loop loop)
CODE:
{
    PERL_UNUSED_VAR(class);
    Newxz(RETVAL, 1, ev_mariadb_t);
    RETVAL->magic = EV_MARIADB_MAGIC;
    RETVAL->loop = loop;
    RETVAL->fd = -1;
    RETVAL->state = STATE_IDLE;
    ngx_queue_init(&RETVAL->cb_queue);
    ngx_queue_init(&RETVAL->send_queue);

    ev_init(&RETVAL->timer, timer_cb);
    RETVAL->timer.data = (void *)RETVAL;
}
OUTPUT:
    RETVAL

void
DESTROY(EV::MariaDB self)
CODE:
{
    if (self->magic != EV_MARIADB_MAGIC)
        return;

    self->magic = EV_MARIADB_FREED;

    stop_reading(self);
    stop_writing(self);
    stop_timer(self);

    if (PL_dirty) {
        /* global destruction — free C resources only; skip SvREFCNT_dec
           to avoid cascading destructors in torn-down interpreter */
        int is_fork = IS_FORKED(self);
        while (!ngx_queue_empty(&self->send_queue)) {
            ngx_queue_t *q = ngx_queue_head(&self->send_queue);
            ev_mariadb_send_t *s = ngx_queue_data(q, ev_mariadb_send_t, queue);
            ngx_queue_remove(q);
            Safefree(s->sql);
            Safefree(s);
        }
        while (!ngx_queue_empty(&self->cb_queue)) {
            ngx_queue_t *q = ngx_queue_head(&self->cb_queue);
            ev_mariadb_cb_t *cbt = ngx_queue_data(q, ev_mariadb_cb_t, queue);
            ngx_queue_remove(q);
            Safefree(cbt);
        }
        if (self->op_data_ptr) Safefree(self->op_data_ptr);
        Safefree(self->pending_user);
        Safefree(self->pending_password);
        Safefree(self->pending_database);
        Safefree(self->pending_charset);
        self->stream_cb = NULL;
        {
            ev_mariadb_stmt_t *ctx = self->stmt_list;
            while (ctx) {
                ev_mariadb_stmt_t *next = ctx->next;
                free_stmt_bind_params(ctx);
                if (!is_fork && ctx->stmt) {
                    if (ctx->stmt == self->op_stmt) self->op_stmt = NULL;
                    mysql_stmt_close(ctx->stmt);
                }
                Safefree(ctx);
                ctx = next;
            }
        }
        if (!is_fork) {
            if (self->op_result) mysql_free_result(self->op_result);
            if (self->conn) mysql_close(self->conn);
        }
        free_connect_strings(self);
        free_option_strings(self);
        Safefree(self);
        return;
    }

    cancel_pending(self, "object destroyed");

    /* safety net: callbacks during cancel_pending could re-queue entries */
    drain_queues_silent(self);

    if (self->op_data_ptr) {
        Safefree(self->op_data_ptr);
        self->op_data_ptr = NULL;
    }

    FREE_STR(self->pending_user);
    FREE_STR(self->pending_password);
    FREE_STR(self->pending_database);
    FREE_STR(self->pending_charset);

    /* Free all tracked stmt wrappers and connection resources */
    {
        int is_fork = IS_FORKED(self);
        ev_mariadb_stmt_t *ctx = self->stmt_list;
        while (ctx) {
            ev_mariadb_stmt_t *next = ctx->next;
            free_stmt_bind_params(ctx);
            if (!is_fork && ctx->stmt) {
                /* op_stmt may equal a tracked stmt (e.g. STATE_STMT_RESET,
                   STATE_STMT_STORE, STATE_STMT_SEND_LONG_DATA where op_stmt_ctx
                   is NULL); avoid the double-close below. */
                if (ctx->stmt == self->op_stmt) self->op_stmt = NULL;
                mysql_stmt_close(ctx->stmt);
            }
            Safefree(ctx);
            ctx = next;
        }
        self->stmt_list = NULL;
        /* Close op_stmt only if it was never linked into stmt_list (in-flight prepare). */
        if (!is_fork && self->op_stmt && !self->op_stmt_ctx) {
            mysql_stmt_close(self->op_stmt);
        }
        self->op_stmt = NULL;
        self->op_stmt_ctx = NULL;

        if (is_fork) {
            self->op_result = NULL;
            self->conn = NULL;
        } else {
            if (self->op_result) {
                mysql_free_result(self->op_result);
                self->op_result = NULL;
            }
            {
                MYSQL *conn = self->conn;
                self->conn = NULL;
                if (conn) mysql_close(conn);
            }
        }
    }

    /* Drop handler SVs before zeroing loop/fd so any re-entrant destructors
       see NULL handler fields (and a still-valid loop while unwinding). */
    {
        SV *cb;
        if ((cb = self->on_connect)) { self->on_connect = NULL; SvREFCNT_dec(cb); }
        if ((cb = self->on_error))   { self->on_error   = NULL; SvREFCNT_dec(cb); }
    }

    self->loop = NULL;
    self->fd = -1;

    free_connect_strings(self);
    free_option_strings(self);

    /* deferred free: check_destroyed will Safefree when depth hits 0 */
    if (self->callback_depth == 0)
        Safefree(self);
}

void
connect(EV::MariaDB self, const char *host, const char *user, const char *password, const char *database, unsigned int port = 3306, SV *unix_socket_sv = NULL)
CODE:
{
    if (NULL != self->conn) {
        croak(self->state == STATE_CONNECTING
              ? "connection already in progress"
              : "already connected");
    }

    free_connect_strings(self);

    self->host = safe_strdup(host);
    self->user = safe_strdup(user);
    self->password = safe_strdup(password);
    self->database = safe_strdup(database);
    self->port = port;
    self->unix_socket = NULL;

    if (unix_socket_sv && SvOK(unix_socket_sv)) {
        self->unix_socket = safe_strdup(SvPV_nolen(unix_socket_sv));
    }

    start_connect(self);
}

void
reset(EV::MariaDB self)
CODE:
{
    if (NULL == self->host) {
        croak("no previous connection to reset");
    }

    cleanup_connection(self);
    cancel_pending(self, "connection reset");
    if (check_destroyed(self)) return;
    if (self->state != STATE_IDLE) return;

    start_connect(self);
}

void
finish(EV::MariaDB self)
CODE:
{
    cleanup_connection(self);
    cancel_pending(self, "connection finished");
    if (check_destroyed(self)) return;
}

SV*
on_connect(EV::MariaDB self, SV *handler = NULL)
CODE:
{
    RETVAL = handler_accessor(&self->on_connect, handler, items > 1);
}
OUTPUT:
    RETVAL

SV*
on_error(EV::MariaDB self, SV *handler = NULL)
CODE:
{
    RETVAL = handler_accessor(&self->on_error, handler, items > 1);
}
OUTPUT:
    RETVAL

void
query(EV::MariaDB self, SV *sql_sv, SV *cb)
CODE:
{
    STRLEN sql_len;
    const char *sql;

    if (NULL == self->conn) {
        croak("not connected");
    }
    if (IS_FORKED(self)) {
        croak("connection not valid after fork");
    }
    if (self->state >= STATE_STMT_PREPARE) {
        croak("cannot queue query: exclusive operation in progress");
    }
    if (!(SvROK(cb) && SvTYPE(SvRV(cb)) == SVt_PVCV)) {
        croak("callback must be a CODE reference");
    }

    sql = SvPV(sql_sv, sql_len);
    push_send(self, sql, sql_len, cb);

    if (self->state == STATE_IDLE && self->callback_depth == 0)
        pipeline_advance(self);
}

void
prepare(EV::MariaDB self, SV *sql_sv, SV *cb)
PREINIT:
    STRLEN sql_len;
    const char *sql;
CODE:
{
    int status;
    MYSQL_STMT *stmt;

    CHECK_READY(self, cb);

    stmt = mysql_stmt_init(self->conn);
    if (NULL == stmt) {
        croak("mysql_stmt_init failed");
    }
    {
        my_bool update_max = 1;
        mysql_stmt_attr_set(stmt, STMT_ATTR_UPDATE_MAX_LENGTH, &update_max);
    }

    sql = SvPV(sql_sv, sql_len);
    push_cb(self, cb);
    self->op_stmt = stmt;

    self->state = STATE_STMT_PREPARE;
    status = mysql_stmt_prepare_start(&self->op_ret, stmt, sql, (unsigned long)sql_len);
    if (status == 0) {
        on_stmt_prepare_done(self);
    } else {
        update_watchers(self, status);
    }
}

void
execute(EV::MariaDB self, IV stmt_iv, SV *params_ref, SV *cb)
PREINIT:
    ev_mariadb_stmt_t *ctx;
    int status;
CODE:
{
    CHECK_READY(self, cb);

    ctx = INT2PTR(ev_mariadb_stmt_t *, stmt_iv);
    CHECK_STMT(ctx);

    if (SvOK(params_ref)) {
        if (!SvROK(params_ref) || SvTYPE(SvRV(params_ref)) != SVt_PVAV)
            croak("params must be an ARRAY reference or undef");
        setup_bind_params(ctx, (AV *)SvRV(params_ref));
    }

    push_cb(self, cb);
    self->op_stmt = ctx->stmt;
    self->op_stmt_ctx = ctx;

    self->state = STATE_STMT_EXECUTE;
    status = mysql_stmt_execute_start(&self->op_ret, ctx->stmt);
    if (status == 0) {
        transition_to_stmt_store(self);
    } else {
        update_watchers(self, status);
    }
}

void
bind_params(EV::MariaDB self, IV stmt_iv, SV *params_ref)
PREINIT:
    ev_mariadb_stmt_t *ctx;
CODE:
{
    if (NULL == self->conn || self->state == STATE_CONNECTING)
        croak("not connected");
    if (IS_FORKED(self))
        croak("connection not valid after fork");
    if (self->state != STATE_IDLE)
        croak("another operation is in progress");
    if (self->send_count > 0)
        croak("cannot bind while pipeline results are pending");
    if (!SvROK(params_ref) || SvTYPE(SvRV(params_ref)) != SVt_PVAV)
        croak("params must be an ARRAY reference");

    ctx = INT2PTR(ev_mariadb_stmt_t *, stmt_iv);
    CHECK_STMT(ctx);
    setup_bind_params(ctx, (AV *)SvRV(params_ref));
}

void
close_stmt(EV::MariaDB self, IV stmt_iv, SV *cb)
PREINIT:
    ev_mariadb_stmt_t *ctx;
    int status;
CODE:
{
    CHECK_READY(self, cb);

    ctx = INT2PTR(ev_mariadb_stmt_t *, stmt_iv);
    if (!ctx) croak("invalid statement handle");
    if (ctx->closed) {
        /* already closed by cleanup_connection — just free the wrapper */
        unlink_stmt(self, ctx);
        Safefree(ctx);
        push_cb(self, cb);
        if (deliver_value(self, newSViv(1))) return;
        maybe_pipeline(self);
        return;
    }
    free_stmt_bind_params(ctx);
    push_cb(self, cb);
    self->op_stmt = ctx->stmt;
    self->op_stmt_ctx = ctx;

    self->state = STATE_STMT_CLOSE;
    status = mysql_stmt_close_start(&self->op_bool_ret, ctx->stmt);
    if (status == 0) {
        on_stmt_close_done(self);
    } else {
        update_watchers(self, status);
    }
}

void
stmt_reset(EV::MariaDB self, IV stmt_iv, SV *cb)
PREINIT:
    ev_mariadb_stmt_t *ctx;
    int status;
CODE:
{
    CHECK_READY(self, cb);

    ctx = INT2PTR(ev_mariadb_stmt_t *, stmt_iv);
    CHECK_STMT(ctx);
    push_cb(self, cb);
    self->op_stmt = ctx->stmt;

    self->state = STATE_STMT_RESET;
    status = mysql_stmt_reset_start(&self->op_bool_ret, ctx->stmt);
    if (status == 0) {
        on_stmt_reset_done(self);
    } else {
        update_watchers(self, status);
    }
}

void
ping(EV::MariaDB self, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    push_cb(self, cb);

    self->state = STATE_PING;
    status = mysql_ping_start(&self->op_ret, self->conn);
    dispatch_utility(self, status, self->op_ret != 0);
}

void
change_user(EV::MariaDB self, const char *user, const char *password, SV *db_sv, SV *cb)
PREINIT:
    int status;
    const char *db;
CODE:
{
    CHECK_READY(self, cb);

    db = (SvOK(db_sv)) ? SvPV_nolen(db_sv) : NULL;

    /* Stash the new credentials; on_utility_done commits them on success or
       discards on failure so reset() never sees credentials we couldn't use.
       Pass the pending pointers to libmariadb so the strings live across the
       async start/cont sequence (Perl args expire when the XS body returns). */
    replace_str(&self->pending_user, user);
    replace_str(&self->pending_password, password);
    if (db) replace_str(&self->pending_database, db);

    push_cb(self, cb);

    self->state = STATE_CHANGE_USER;
    status = mysql_change_user_start(&self->op_bool_ret, self->conn,
        self->pending_user, self->pending_password, self->pending_database);
    dispatch_utility(self, status, self->op_bool_ret != 0);
}

void
select_db(EV::MariaDB self, const char *db, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    /* Stash the new database; on_utility_done commits on success. */
    replace_str(&self->pending_database, db);

    push_cb(self, cb);

    self->state = STATE_SELECT_DB;
    status = mysql_select_db_start(&self->op_ret, self->conn, self->pending_database);
    dispatch_utility(self, status, self->op_ret != 0);
}

void
reset_connection(EV::MariaDB self, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    push_cb(self, cb);

    self->state = STATE_RESET_CONNECTION;
    status = mysql_reset_connection_start(&self->op_ret, self->conn);
    dispatch_utility(self, status, self->op_ret != 0);
}

void
set_charset(EV::MariaDB self, const char *charset, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    /* Stash the new charset; on_utility_done commits on success. */
    replace_str(&self->pending_charset, charset);

    push_cb(self, cb);

    self->state = STATE_SET_CHARSET;
    status = mysql_set_character_set_start(&self->op_ret, self->conn, self->pending_charset);
    dispatch_utility(self, status, self->op_ret != 0);
}

void
commit(EV::MariaDB self, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    push_cb(self, cb);

    self->state = STATE_COMMIT;
    status = mysql_commit_start(&self->op_bool_ret, self->conn);
    dispatch_utility(self, status, self->op_bool_ret != 0);
}

void
rollback(EV::MariaDB self, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    push_cb(self, cb);

    self->state = STATE_ROLLBACK;
    status = mysql_rollback_start(&self->op_bool_ret, self->conn);
    dispatch_utility(self, status, self->op_bool_ret != 0);
}

void
autocommit(EV::MariaDB self, int mode, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    push_cb(self, cb);

    self->state = STATE_AUTOCOMMIT;
    status = mysql_autocommit_start(&self->op_bool_ret, self->conn, (my_bool)(mode ? 1 : 0));
    dispatch_utility(self, status, self->op_bool_ret != 0);
}

void
query_stream(EV::MariaDB self, SV *sql_sv, SV *cb)
PREINIT:
    STRLEN sql_len;
    const char *sql;
    int status;
CODE:
{
    CHECK_READY(self, cb);

    sql = SvPV(sql_sv, sql_len);
    self->stream_cb = SvREFCNT_inc(cb);
    self->pending_count++;

    self->state = STATE_REAL_QUERY;
    status = mysql_real_query_start(&self->op_ret, self->conn, sql, (unsigned long)sql_len);
    if (status == 0) {
        on_real_query_done(self);
    } else {
        update_watchers(self, status);
    }
}

void
close_async(EV::MariaDB self, SV *cb)
PREINIT:
    int status;
CODE:
{
    CHECK_READY(self, cb);

    push_cb(self, cb);

    self->state = STATE_CLOSE;
    status = mysql_close_start(self->conn);
    if (status == 0) {
        on_close_done(self);
    } else {
        update_watchers(self, status);
    }
}

void
send_long_data(EV::MariaDB self, IV stmt_iv, unsigned int param_idx, SV *data_sv, SV *cb)
PREINIT:
    ev_mariadb_stmt_t *ctx;
    STRLEN data_len;
    char *data;
    int status;
CODE:
{
    CHECK_READY(self, cb);

    ctx = INT2PTR(ev_mariadb_stmt_t *, stmt_iv);
    CHECK_STMT(ctx);
    {
        const char *src = SvPV(data_sv, data_len);
        Newx(data, data_len > 0 ? data_len : 1, char);
        Copy(src, data, data_len, char);
    }
    push_cb(self, cb);
    self->op_stmt = ctx->stmt;
    self->op_data_ptr = data;

    self->state = STATE_STMT_SEND_LONG_DATA;
    status = mysql_stmt_send_long_data_start(&self->op_bool_ret, ctx->stmt,
        param_idx, data, (unsigned long)data_len);
    if (status == 0) {
        on_send_long_data_done(self);
    } else {
        update_watchers(self, status);
    }
}

void
_set_option(EV::MariaDB self, const char *key, SV *value)
CODE:
{
    if (strcmp(key, "connect_timeout") == 0) {
        self->connect_timeout = SvUV(value);
    } else if (strcmp(key, "read_timeout") == 0) {
        self->read_timeout = SvUV(value);
    } else if (strcmp(key, "write_timeout") == 0) {
        self->write_timeout = SvUV(value);
    } else if (strcmp(key, "compress") == 0) {
        self->compress = SvTRUE(value) ? 1 : 0;
    } else if (strcmp(key, "multi_statements") == 0) {
        self->multi_statements = SvTRUE(value) ? 1 : 0;
    } else if (strcmp(key, "found_rows") == 0) {
        self->found_rows = SvTRUE(value) ? 1 : 0;
    } else if (strcmp(key, "charset") == 0) {
        SET_STR_OPTION(charset);
    } else if (strcmp(key, "init_command") == 0) {
        SET_STR_OPTION(init_command);
    } else if (strcmp(key, "ssl_key") == 0) {
        SET_STR_OPTION(ssl_key);
    } else if (strcmp(key, "ssl_cert") == 0) {
        SET_STR_OPTION(ssl_cert);
    } else if (strcmp(key, "ssl_ca") == 0) {
        SET_STR_OPTION(ssl_ca);
    } else if (strcmp(key, "ssl_capath") == 0) {
        SET_STR_OPTION(ssl_capath);
    } else if (strcmp(key, "ssl_cipher") == 0) {
        SET_STR_OPTION(ssl_cipher);
    } else if (strcmp(key, "ssl_verify_server_cert") == 0) {
        self->ssl_verify_server_cert = SvTRUE(value) ? 1 : 0;
    } else if (strcmp(key, "utf8") == 0) {
        self->utf8 = SvTRUE(value) ? 1 : 0;
    } else {
        croak("unknown option: %s", key);
    }
}

int
is_connected(EV::MariaDB self)
CODE:
{
    RETVAL = (NULL != self->conn && self->state != STATE_CONNECTING) ? 1 : 0;
}
OUTPUT:
    RETVAL

SV*
error_message(EV::MariaDB self)
CODE:
{
    RETVAL = self->conn ? str_or_undef(mysql_error(self->conn)) : &PL_sv_undef;
}
OUTPUT:
    RETVAL

unsigned int
error_number(EV::MariaDB self)
CODE:
{
    RETVAL = (NULL != self->conn) ? mysql_errno(self->conn) : 0;
}
OUTPUT:
    RETVAL

SV*
sqlstate(EV::MariaDB self)
CODE:
{
    RETVAL = self->conn ? str_or_undef(mysql_sqlstate(self->conn)) : &PL_sv_undef;
}
OUTPUT:
    RETVAL

SV*
insert_id(EV::MariaDB self)
CODE:
{
    if (NULL != self->conn) {
        my_ulonglong id = mysql_insert_id(self->conn);
        RETVAL = newSVuv((UV)id);
    }
    else {
        RETVAL = &PL_sv_undef;
    }
}
OUTPUT:
    RETVAL

unsigned int
warning_count(EV::MariaDB self)
CODE:
{
    RETVAL = (NULL != self->conn) ? mysql_warning_count(self->conn) : 0;
}
OUTPUT:
    RETVAL

SV*
info(EV::MariaDB self)
CODE:
{
    RETVAL = self->conn ? str_or_undef(mysql_info(self->conn)) : &PL_sv_undef;
}
OUTPUT:
    RETVAL

UV
server_version(EV::MariaDB self)
CODE:
{
    RETVAL = (NULL != self->conn) ? (UV)mysql_get_server_version(self->conn) : 0;
}
OUTPUT:
    RETVAL

SV*
server_info(EV::MariaDB self)
CODE:
{
    RETVAL = self->conn ? str_or_undef(mysql_get_server_info(self->conn)) : &PL_sv_undef;
}
OUTPUT:
    RETVAL

UV
thread_id(EV::MariaDB self)
CODE:
{
    RETVAL = (NULL != self->conn) ? (UV)mysql_thread_id(self->conn) : 0;
}
OUTPUT:
    RETVAL

SV*
host_info(EV::MariaDB self)
CODE:
{
    RETVAL = self->conn ? str_or_undef(mysql_get_host_info(self->conn)) : &PL_sv_undef;
}
OUTPUT:
    RETVAL

SV*
character_set_name(EV::MariaDB self)
CODE:
{
    RETVAL = self->conn ? str_or_undef(mysql_character_set_name(self->conn)) : &PL_sv_undef;
}
OUTPUT:
    RETVAL

int
socket(EV::MariaDB self)
CODE:
{
    RETVAL = (NULL != self->conn) ? (int)mysql_get_socket(self->conn) : -1;
}
OUTPUT:
    RETVAL

SV*
affected_rows(EV::MariaDB self)
CODE:
{
    if (NULL != self->conn) {
        my_ulonglong rows = mysql_affected_rows(self->conn);
        RETVAL = (rows == (my_ulonglong)-1) ? &PL_sv_undef : newSVuv((UV)rows);
    }
    else {
        RETVAL = &PL_sv_undef;
    }
}
OUTPUT:
    RETVAL

SV*
escape(EV::MariaDB self, SV *str)
PREINIT:
    STRLEN len;
    const char *s;
    unsigned long elen;
CODE:
{
    if (NULL == self->conn || self->state == STATE_CONNECTING || IS_FORKED(self))
        croak("not connected");
    if (self->state == STATE_CLOSE)
        croak("connection is closing");
    s = SvPV(str, len);
    if (SvUTF8(str)) {
        const char *cs = mysql_character_set_name(self->conn);
        if (!cs || (strncmp(cs, "utf8", 4) != 0)) {
            warn("EV::MariaDB: escaping a UTF-8 string on a non-utf8 connection (%s) may cause corruption or injection vulnerabilities", cs ? cs : "unknown");
        }
    }
    if (len > (STRLEN)((~(STRLEN)0) / 2 - 1))
        croak("string too large to escape");
    RETVAL = newSV(len * 2 + 1);
    SvPOK_on(RETVAL);
    elen = mysql_real_escape_string(self->conn, SvPVX(RETVAL), s, (unsigned long)len);
    if (elen == (unsigned long)-1) {
        SvREFCNT_dec(RETVAL);
        croak("mysql_real_escape_string failed");
    }
    SvCUR_set(RETVAL, elen);
    *SvEND(RETVAL) = '\0';
}
OUTPUT:
    RETVAL

int
pending_count(EV::MariaDB self)
CODE:
{
    RETVAL = self->pending_count;
}
OUTPUT:
    RETVAL

void
skip_pending(EV::MariaDB self)
CODE:
{
    if (self->state != STATE_IDLE || self->send_count > 0) {
        cleanup_connection(self);
    }
    cancel_pending(self, "skipped");
    if (check_destroyed(self)) return;
}

UV
lib_version(char *class)
CODE:
{
    PERL_UNUSED_VAR(class);
    RETVAL = (UV)mysql_get_client_version();
}
OUTPUT:
    RETVAL

SV*
lib_info(char *class)
CODE:
{
    PERL_UNUSED_VAR(class);
    RETVAL = newSVpv(mysql_get_client_info(), 0);
}
OUTPUT:
    RETVAL