view release on metacpan or  search on metacpan

include/pq_backend.h  view on Meta::CPAN

    pq_txn_begin(aTHX_ self);
    if (!pq_guarded_peek(aTHX_ self, id, retries, prestate, sizeof prestate,
                         &attempts_col)) {
        pq_txn_commit(aTHX_ self);
        return 0;
    }
    now = pq_now(aTHX_ self);

    sql = pq_sql_new(aTHX_
        "UPDATE pq_jobs SET state = 'inactive', retries = retries + 1,"
        " retried = ?, delayed = ?, started = NULL, finished = NULL,"
        " worker = NULL, result = NULL");
    if (opts && (v = pq_get(aTHX_ opts, "delay")) && SvOK(v))
        delay = SvNV(v);
    pq_bind_nv(aTHX_ bind, now);
    pq_bind_nv(aTHX_ bind, now + delay);

    if (opts && (v = pq_get(aTHX_ opts, "priority")) && SvOK(v)) {
        pq_sql_cat(aTHX_ sql, ", priority = ?");
        pq_bind_iv(aTHX_ bind, SvIV(v));
    }
    if (opts && (v = pq_get(aTHX_ opts, "queue")) && SvOK(v)) {
        pq_name_check(aTHX_ v, "queue");
        pq_sql_cat(aTHX_ sql, ", queue = ?");
        pq_bind_sv(aTHX_ bind, v);
    }
    if (opts && (v = pq_get(aTHX_ opts, "attempts")) && SvOK(v)) {
        if (SvIV(v) < 1) {
            pq_txn_rollback(aTHX_ self);
            croak("Punk::Queue: attempts must be at least 1, not %ld",
                  (long)SvIV(v));
        }
        pq_sql_cat(aTHX_ sql, ", attempts = ?");
        pq_bind_iv(aTHX_ bind, SvIV(v));
    }

    pq_sql_cat(aTHX_ sql, " WHERE id = ? AND retries = ?");
    pq_bind_iv(aTHX_ bind, id);
    pq_bind_iv(aTHX_ bind, retries);

    took = pq_do(aTHX_ self, sql, bind);
    if (took > 0) {
        /* the children had counted this parent done; block them again */
        if (strEQ(prestate, "finished"))
            pq_children_adjust(aTHX_ self, id, +1, PQ_KIDS_ALL);
        else if (strEQ(prestate, "failed"))
            pq_children_adjust(aTHX_ self, id, +1, PQ_KIDS_LAX);
        pq_log_addf(aTHX_ self, id, "info",
                    delay > 0 ? "retried from '%s' - due in %.1" NVff "s"
                              : "retried from '%s'",
                    prestate, (NV)delay);
    }
    pq_txn_commit(aTHX_ self);
    return took > 0 ? 1 : 0;
}

/* remove_job refuses an active job: the worker running it would finish
 * into a void and the supervision bookkeeping would point at a ghost.
 * Retry it first (which strands the running attempt), then remove. Dep
 * rows cascade via the foreign keys - but only after the children's
 * counters are adjusted, because the adjustment needs the dep rows to
 * find them. */
static IV pq_remove_job(pTHX_ SV *self, IV id) {
    char prestate[16];
    IV attempts = 0, took = 0;

    pq_txn_begin(aTHX_ self);
    {
        AV *bind = pq_binds(aTHX), *row;
        pq_bind_iv(aTHX_ bind, id);
        row = pq_selectrow(aTHX_ self, sv_2mortal(newSVpvs(
            "SELECT state, attempts FROM pq_jobs WHERE id = ?")), bind);
        if (!row) { pq_txn_commit(aTHX_ self); return 0; }
        {
            SV *st = pq_col(aTHX_ row, 0);
            my_strlcpy(prestate, SvOK(st) ? SvPV_nolen(st) : "",
                       sizeof prestate);
            attempts = pq_col_iv(aTHX_ row, 1);
        }
        SvREFCNT_dec((SV *)row);
        PERL_UNUSED_VAR(attempts);
    }
    if (strEQ(prestate, "active")) {
        pq_txn_commit(aTHX_ self);
        return 0;
    }

    if (strEQ(prestate, "inactive"))
        pq_children_adjust(aTHX_ self, id, -1, PQ_KIDS_ALL);
    else if (strEQ(prestate, "failed"))
        pq_children_adjust(aTHX_ self, id, -1, PQ_KIDS_NONLAX);
    /* finished: every child already got its decrement */

    {
        AV *bind = pq_binds(aTHX);
        pq_bind_iv(aTHX_ bind, id);
        took = pq_do(aTHX_ self, sv_2mortal(newSVpvs(
            "DELETE FROM pq_jobs WHERE id = ?"
            " AND state IN ('inactive', 'failed', 'finished')")), bind);
    }
    if (took > 0) {
        AV *bind = pq_binds(aTHX);
        pq_bind_iv(aTHX_ bind, id);
        (void)pq_do(aTHX_ self, sv_2mortal(newSVpvs(
            "DELETE FROM pq_job_logs WHERE job_id = ?")), bind);
    }
    pq_txn_commit(aTHX_ self);
    return took > 0 ? 1 : 0;
}

/* ---- notes -----------------------------------------------------------------
 *
 * Merge, not replace, and merged in C after an frj decode rather than in
 * SQL - json_patch exists on neither backend in a portable form, and the
 * conformance suite requires byte-identical behaviour. A key set to undef
 * is deleted. Works in any state: progress notes on an active job are the
 * whole point. */
static IV pq_note(pTHX_ SV *self, IV id, HV *merge) {
    AV *bind, *row;
    SV *decoded, *json;