NAME
Langertha::Knarr::TokenStore::Pg - PostgreSQL token store for token_auth, asynchronous under IO::Async
VERSION
version 1.103
SYNOPSIS
use Langertha::Knarr::TokenStore::Pg;
my $store = Langertha::Knarr::TokenStore::Pg->new(
dsn => 'dbi:Pg:host=db;dbname=platform',
user => 'tester_knarr',
password_env => 'TESTER_DB_PASSWORD',
relation => 'citiai_tester_token_auth', # or schema.view
listen => 'citiai_tester_token',
columns => { hash => 'token_hash', model => 'model',
provider => 'provider', expires => 'expires_at' },
loop => $loop,
);
$store->on_invalidate( sub { %cache = () } );
my $row = await $store->lookup_f($hash_hex); # in the event loop
my $row = $store->lookup($hash_hex); # blocking (PSGI)DESCRIPTION
The Langertha::Knarr::TokenStore of token_auth with backend: postgresql. It looks a token hash up in a table or view the platform that issues the tokens owns, read-only:
SELECT <model>, <provider>, extract(epoch from <expires>)
FROM <relation> WHERE <hash> = $1 AND <expires> > now()The relation (schema.view or a bare name; a dot separates the schema) and the column names are quoted as identifiers, so they are taken as they are written, case included. The hash column holds the lowercase hex SHA-256 of the token; the expiry column should be a timestamptz.
The store LISTENs on "listen". Every notification on that channel -- the platform sends one when it issues or revokes a token -- and every loss of the connection calls "on_invalidate" in Langertha::Knarr::TokenStore, so the caller's cache never outlives a revocation it could not see.
DBD::Pg is an optional dependency of Knarr. Constructing this store without it croaks token_auth backend postgresql needs DBD::Pg, so knarr start fails at once instead of on the first request.
Asynchronous queries
A lookup sits in the path of every request whose token is not cached, so it must not stop the event loop while the database answers. The store uses DBD::Pg's own asynchronous mode: the query is sent with pg_async => PG_ASYNC and returns at once, and the connection's socket (pg_socket) is watched by an IO::Async::Handle. When it turns readable the store collects the result (pg_ready, pg_result) and the notifications (pg_notifies) without waiting. The same watch delivers NOTIFY the moment it arrives -- a revocation reaches the cache without any polling -- and it sees the connection end the moment the server closes it.
The alternatives were worse for this job. A forked helper process with a blocking DBI handle needs its own protocol and process supervision, and it would sit between the LISTEN and the cache, delaying every invalidation. A blocking query in the loop stops every request of the worker, streaming ones included, for as long as the database takes. The asynchronous mode keeps one connection per serving process, under the loop the server already runs.
One connection runs one asynchronous query at a time, so lookups wait in a queue; a lookup for a hash already queued or running joins that query instead of sending its own. A query that takes longer than "query_timeout" counts as a lost connection.
DBD::Pg has no asynchronous connect, and a DBI connect to a server that is down or unreachable would stop the loop for its whole timeout -- on every reconnect attempt, with streams running. Under the loop the store therefore first opens a plain TCP connection to the DSN's host (hostaddr, else PGHOST) and port (else PGPORT, else 5432) through the loop, without blocking, for at most "connect_timeout" seconds. Only when that gets through does it call DBI->connect; a failed probe is a failed attempt and goes to the backoff without any DBI call. Lookups that arrive while an attempt runs wait for it; lookups during the backoff fail at once. The probe connection is closed right away (the server may log it as an incomplete startup packet). A DSN without a TCP host -- a Unix socket directory, service=, a list of hosts -- is not probed: its connect blocks the loop for up to "connect_timeout".
The DSN gets connect_timeout ("connect_timeout") and TCP keepalives (keepalives_idle=30, keepalives_interval=10, keepalives_count=3) unless it sets them itself, so neither the connect after a probe nor an idle connection to a vanished server hangs for longer than that.
Connection loss and processes
The connection is opened by "start" -- which Langertha::Knarr calls in the process that serves -- or else by the first lookup: with workers above 1 every worker opens its own connection and LISTEN. A connection inherited across a fork is noticed by the process id, left to its owner untouched (InactiveDestroy) and replaced; the caller's cache is invalidated then too, since the notifications since the fork went to the other process.
When the connection is lost (the server closes it, a query fails with a connection error, "query_timeout"), the store is not "available", calls the invalidate listener with disconnect, fails every queued lookup and connects again after 1, 2, 4, ... seconds, at most 30, with a new LISTEN. Until then every lookup fails at once.
Secrets
The password is read from the environment variable named by "password_env" at every connect and is never stored, logged or put in an error. A failed lookup fails with the fixed text token store unavailable; database errors (which name the host and user) go only to the log, first line only, with the password and any token hash masked.
dsn
Required. The DBI data source, dbi:Pg:....
user
Optional database user.
password_env
Optional name of the environment variable that holds the database password. Without it libpq's own sources (.pgpass, PGPASSWORD) apply.
relation
Required. The table or view to look tokens up in, name or schema.name.
listen
Required. The channel to LISTEN on; every notification on it invalidates.
columns
HashRef of the column names, hash, model, provider, expires. Missing entries default to token_hash, model, provider, expires_at.
loop
Optional IO::Async::Loop for "lookup_f". Defaults to IO::Async::Loop->new.
connect_timeout
Seconds a connect may take: the non-blocking TCP probe under the loop, and libpq's connect_timeout (libpq takes at least 2). Default 3. A connect_timeout in the DSN wins for libpq; the probe still uses this value.
query_timeout
Seconds an asynchronous lookup may take before the connection counts as lost. Default 10.
available
True while the store has a connection of this process with its LISTEN in place. False before the first connect: "lookup_f" and "lookup" connect then, so a caller must not refuse on this alone before it asked.
start
$store->start;Starts a connect attempt (probe, then connect; see "Asynchronous queries") when there is no connection and no reconnect is pending, and switches the store to the loop: from then on its socket is watched and a lost connection is retried by a timer. It returns before the attempt is done. A second call while connected, connecting or waiting to reconnect does nothing. Langertha::Knarr calls it (through "start" in Langertha::Knarr::Authorizer::Token) in every process that serves, once its loop is set up; without that the first "lookup_f" does the same.
lookup_f
my $row = await $store->lookup_f($hash_hex);See "lookup_f" in Langertha::Knarr::TokenStore. Fails with token store unavailable while there is no connection, after a database error and when the connection is lost before the answer.
lookup
my $row = $store->lookup($hash_hex);Blocking, for Langertha::Knarr::PSGI: delivers pending notifications first (as "poll"), then queries. It connects without the TCP probe, so a connect may block for up to "connect_timeout". Croaks token store unavailable on any store error. Without a loop a lost connection is retried by the next lookup once the backoff has passed. A store serves either the loop ("lookup_f") or blocking callers, not both.
poll
$store->poll;Delivers the notifications that have arrived without waiting, and notices a closed connection. A blocking caller calls it before it trusts its cache; under the loop the socket watch does this by itself.
SUPPORT
Issues
Please report bugs and feature requests on GitHub at https://github.com/Getty/langertha-knarr/issues.
IRC
Join #langertha on irc.perl.org or message Getty directly.
CONTRIBUTING
Contributions are welcome! Please fork the repository and submit a pull request.
AUTHOR
Torsten Raudssus <torsten@raudssus.de> https://raudssus.de/
COPYRIGHT AND LICENSE
This software is copyright (c) 2026 by Torsten Raudssus.
This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.