Punk-Queue
view release on metacpan or search on metacpan
example/Mailer/README.pod view on Meta::CPAN
C<Job::Report::signups> holds a named lock for the duration. Locks live in
the database with the jobs, so it holds across every worker on every host,
and taking it with a duration means a crashed holder releases itself.
=head2 The admin UI is yours to restyle
F<root/queue/> is a template root searched before the bundled one, one name
at a time - it holds exactly the one template this app changes, and the
other six keep coming from the dist across upgrades. F<root/queue.css> is
appended to the served bundle, after Funky and after C<punk-queue.css>, so
the cascade takes it. Both are read once, at registration: a template edit
needs a restart, like everything else in a Punk app.
Which root serves which template is decided at registration too, so an
override you meant to make but misspelled is a boot croak rather than a
page that silently stays ours.
=head2 In-server mode is a shortcut, and knows it
MAILER_IN_SERVER=1 plackup -s Hyperman -p 5000 app.psgi
example/Mailer/root/queue.css view on Meta::CPAN
/* admin => { css => 'root/queue.css' }
*
* Appended to the served /queue/assets/funky.css, after Funky and after
* punk-queue.css - which is what makes it an override rather than a
* competing sheet: the cascade takes the last declaration.
*
* The bundle is assembled once, at registration, so this file is read at
* boot and never again; its bytes go into the bundle's ETag, so a browser
* holding the un-overridden version is told to refetch.
*/
.pq-brand {
color: #6f4bd8;
letter-spacing: 0.02em;
}
include/pq_backend.h view on Meta::CPAN
: "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);
include/pq_log.h view on Meta::CPAN
*
* Two writers: the transitions log their own lifecycle - claimed,
* finished, attempt failed with the retry it scheduled, terminal
* failure, operator retry - and the task body writes whatever it wants
* through $job->log. The lifecycle rows are gated on the `logging`
* backend option (default on) so a throughput-sensitive queue can shed
* them; an explicit $job->log always writes, because a call the task
* author typed is not noise.
*
* Rows live and die with their job: remove_job and repair's ancient
* sweep cascade here, repair also sweeps orphans, and reset empties the
* table. Retention is therefore the job's retention - inspecting a
* finished job shows its whole story across every attempt, until
* remove_after reaps the job itself.
*
* Include after pq_job.h (uses its hv helpers), before pq_backend.h
* (whose transitions call these). */
static const char *pq_log_level(pTHX_ const char *l) {
if (!l || !*l) return "info";
if (strEQ(l, "debug") || strEQ(l, "info")
include/pq_migrate_pg.h view on Meta::CPAN
,
/* ---- 2 ------------------------------------------------------------------
* In lockstep with PQ_SQLITE_UP step 2. */
"ALTER TABLE pq_jobs ADD COLUMN timeout DOUBLE PRECISION\n"
,
/* ---- 3 ------------------------------------------------------------------
* Per-job log lines: the worker's lifecycle events and whatever the task
* body writes through $job->log. Rows live and die with their job -
* remove_job, repair's ancient sweep and reset all cascade here. */
"CREATE TABLE pq_job_logs (\n"
" id BIGSERIAL PRIMARY KEY,\n"
" job_id BIGINT NOT NULL,\n"
" created DOUBLE PRECISION NOT NULL,\n"
" level TEXT NOT NULL DEFAULT 'info',\n"
" message TEXT NOT NULL\n"
")\n"
"-- @@\n"
"CREATE INDEX pq_job_logs_job ON pq_job_logs (job_id, id)\n"
,
include/pq_migrate_sqlite.h view on Meta::CPAN
/* ---- 2 ------------------------------------------------------------------
* Per-job timeout, seconds; NULL means none. A new column rather than a
* notes key because the supervisor's hard-kill check reads it on every
* claim line and phase 8's UI filters on it. */
"ALTER TABLE pq_jobs ADD COLUMN timeout REAL\n"
,
/* ---- 3 ------------------------------------------------------------------
* Per-job log lines: the worker's lifecycle events and whatever the task
* body writes through $job->log. Rows live and die with their job -
* remove_job, repair's ancient sweep and reset all cascade here. */
"CREATE TABLE pq_job_logs (\n"
" id INTEGER PRIMARY KEY AUTOINCREMENT,\n"
" job_id INTEGER NOT NULL,\n"
" created REAL NOT NULL,\n"
" level TEXT NOT NULL DEFAULT 'info',\n"
" message TEXT NOT NULL\n"
")\n"
"-- @@\n"
"CREATE INDEX pq_job_logs_job ON pq_job_logs (job_id, id)\n"
,
include/pq_repair.h view on Meta::CPAN
for (i = 0; i < nv; i++)
(void)pq_remove_job(aTHX_ self,
SvIV(*av_fetch(victims, i, 0)));
(void)hv_stores(out, "expired_jobs", newSViv((IV)nv));
}
/* 4: ancient terminal jobs are history, not queue. Straight DELETE -
* a finished parent's children got their decrements long ago, and
* failed parents this old have failed children right next to them
* (pass 5 already ran on them in an earlier repair). Dep rows
* cascade. */
{
AV *b = pq_binds(aTHX);
pq_bind_nv(aTHX_ b, now - remove_after);
(void)pq_do(aTHX_ self, sv_2mortal(newSVpvs(
"DELETE FROM pq_job_logs WHERE job_id IN"
" (SELECT id FROM pq_jobs"
" WHERE state IN ('finished', 'failed') AND finished < ?)")),
b);
b = pq_binds(aTHX);
pq_bind_nv(aTHX_ b, now - remove_after);
include/pq_repair.h view on Meta::CPAN
/* 6: stale locks. Reads honour expiry everywhere already; this is
* the table-size half. */
{
AV *b = pq_binds(aTHX);
pq_bind_nv(aTHX_ b, now);
n = pq_do(aTHX_ self, sv_2mortal(newSVpvs(
"DELETE FROM pq_locks WHERE expires <= ?")), b);
(void)hv_stores(out, "stale_locks", newSViv(n));
}
/* 7: orphaned log lines. Every in-tree delete path cascades, so a
* hit here means something ELSE deleted job rows (manual SQL, an
* older binary) - which is exactly the drift repair exists to mop
* up. */
{
n = pq_do(aTHX_ self, sv_2mortal(newSVpvs(
"DELETE FROM pq_job_logs WHERE job_id NOT IN"
" (SELECT id FROM pq_jobs)")), NULL);
(void)hv_stores(out, "orphaned_logs", newSViv(n));
}
lib/Punk/Plugin/Queue/assets/funky/css/03-buttons.css view on Meta::CPAN
/* ============================================
BUTTONS.CSS - Funky Button System
Unified button system with proper cascade order
============================================ */
/* ============================================
CSS VARIABLES - Button Theming
============================================ */
:root {
/* Button Font Sizes (rem for user font-scale support) */
--btn-font-xs: 0.6875rem; /* 11px at 1x scale */
--btn-font-sm: 0.75rem; /* 12px at 1x scale */
--btn-font-md: 0.8125rem; /* 13px at 1x scale */
lib/Punk/Queue/Backend/SQLite.pm view on Meta::CPAN
Suited to a single host: one process or several, one machine. For a queue
shared across machines, use the PostgreSQL backend.
=head2 Connection setup
Every connection gets four pragmas, applied at connect:
journal_mode = WAL readers must not block the claim transaction
synchronous = NORMAL durable enough with WAL, much faster
busy_timeout = 5000 SQLITE_BUSY becomes a wait, not an error
foreign_keys = ON or the dependency cascade silently does nothing
WAL is not optional. C<migrate> refuses to run without it, naming the
reason, because the failure it prevents - workers contending badly under
load - shows up far from its cause. An in-memory database is exempt: it
cannot do WAL and is single-process by definition.
=head2 How a job is claimed
BEGIN IMMEDIATE TRANSACTION
SELECT ... WHERE state = 'inactive' AND queue IN (...)
t/01-migrate.t view on Meta::CPAN
# WAL is mandatory for a file-backed queue: a reader must not block the
# claim transaction. It is applied at connect, and migrate refuses without
# it - this asserts the connect half took.
{
my ($q) = make_queue();
my ($mode) = $q->dbh->selectrow_array('PRAGMA journal_mode');
is(lc $mode, 'wal', 'journal_mode is WAL on a file-backed queue');
my ($fk) = $q->dbh->selectrow_array('PRAGMA foreign_keys');
ok($fk, 'foreign_keys is on, so the dependency cascade works');
}
# The clock delta: probed once per connection, and small against a local
# SQLite (its now() and ours are the same clock).
{
my ($q) = make_queue();
my $delta = $q->backend->clock_delta;
ok(abs($delta) < 5, "clock delta is small for a local database ($delta)");
# bracketed with time() on both sides: a stall between two calls on a
t/77-admin-templates.t view on Meta::CPAN
js => [ File::Spec->catfile($adir, 'extra.js'),
File::Spec->catfile($adir, 'more.js') ] },
};
}
my $aapp = AssetApp2->to_app;
{
my $css = get($aapp, '/q/assets/funky.css')->[2][0];
like($css, qr/rebeccapurple/, 'a css override is served');
ok(index($css, 'rebeccapurple') > index($css, '--pq-'),
'...at the end, after our own sheet, so the cascade favours it');
my $js = get($aapp, '/q/assets/app.js')->[2][0];
like($js, qr/window\.__MINE = 1/, 'a js override is served');
ok(index($js, 'window.__MINE') < index($js, 'window.__ALSO'),
'...in the order given');
ok(index($js, 'window.__MINE') > index($js, 'Funky'),
'...after app.js, which it can therefore build on');
}
# ---- the ETag follows the bytes ----------------------------------------------
( run in 1.025 second using v1.01-cache-2.11-cpan-e7c6538aa59 )