\Drupal\strata\Journal RedisJournal

A journal backed by a Redis stream.

The database journal is always available and is the default. This one exists for the case that breaks it: a site whose write volume makes the journal table itself a contended resource. Two thirds of a Drupal site's captured operations are user access-timestamp churn, so a busy site appends tens of thousands of rows an hour to one table, and every one of those appends sits inside a web request.

A stream, not a list. XADD assigns a monotonic id server-side, which is the same property the table's serial column provides and the reason two concurrent requests get a total order without either taking a lock. A list would need a separate counter and a round trip to read it.

Ids are mapped to sequences, because the interface promises an integer sequence. A Redis stream id is <milliseconds>-<counter>, which does not fit an int without losing the counter. The counter is what orders two appends inside the same millisecond, so it cannot be dropped. The mapping keeps both: the sequence is milliseconds * 1000 + counter, which is monotonic as long as fewer than a thousand operations land in one millisecond - and at a thousand appends a millisecond the journal is not the bottleneck.

Redis is not durable the way a database table is. An XADD acknowledged before the append-only file is fsynced is lost on a hard kill. That is a real trade and it is the operator's to make: the loss window is bounded by Redis's own appendfsync setting, and it is why this is not the default.

Summary

Methods
Properties
Constants
__construct
append
read
trim
pending
pendingBytes
oldest
clear
isSupported
sequenceOf
No public properties found
STREAM
COUNTERS_PER_MILLISECOND
FIELD_OP
FIELD_PAYLOAD
No protected methods found
No protected properties found
No protected constants found
entries
redis
stream
No private constants found

Constant

STREAM

STREAM = 'strata:journal'

Default stream key.

COUNTERS_PER_MILLISECOND

COUNTERS_PER_MILLISECOND = 1000

Counters packed into one millisecond of stream id.

The multiplier that turns a stream id into the integer sequence the interface promises.

FIELD_OP

FIELD_OP = 'op'

Field name the operation is stored under.

FIELD_PAYLOAD

FIELD_PAYLOAD = 'payload'

Field name the payload is stored under.

Properties

$redis

$redis : object

Type

object

$stream

$stream : string

Type

string

Methods

__construct()

__construct(object  $redis, string  $stream = \self::STREAM) : mixed

Constructs a journal.

Parameters

object $redis

A connected client exposing the stream commands: xAdd, xRange, xLen, xDel and del. Typed as an object rather than Redis so this class does not require ext-redis to be loaded to be autoloaded, and so a test can drive it over a stand-in.

string $stream

The stream key.

Throws

\RuntimeException

When the client does not expose the stream commands, which is what an older Redis client or a Predis instance looks like. Refusing here beats discovering it on the first capture.

Returns

mixed —

append()

append(\Drupal\strata\Journal\JournalOp  $operation, ?string  $payload = null) : \Drupal\strata\Journal\JournalOp

Appends an operation and its payload.

Parameters

\Drupal\strata\Journal\JournalOp $operation

The operation to append.

?string $payload

The value the operation captured, or NULL for an operation that carries none.

Returns

\Drupal\strata\Journal\JournalOp —

The operation as appended, carrying its assigned sequence.

read()

read(int  $limit = 5000) : list<array{operation: \Drupal\strata\Journal\JournalOp, payload: string|null}>

Reads a window of operations in capture order, oldest first.

Parameters

int $limit

Most operations to return.

Returns

list

Operations paired with their payloads.

trim()

trim(int  $throughSequence) : int

Removes every operation up to and including a sequence.

Parameters

int $throughSequence

The highest sequence to remove.

Returns

int —

How many operations were removed.

pending()

pending() : int

How many operations are waiting.

Returns

int —

The pending count.

pendingBytes()

pendingBytes() : int

Bytes of payload waiting.

Returns

int —

The pending payload total.

oldest()

oldest() : int|null

Unix microseconds of the oldest waiting operation.

Returns

int|null —

The microtime, or NULL when nothing is waiting.

clear()

clear() : int

Removes everything without flushing it.

Returns

int —

How many operations were removed.

isSupported()

isSupported() : bool

Whether a Redis journal can be used on this host.

Returns

bool —

TRUE when the extension is loaded.

sequenceOf()

sequenceOf(string  $id) : int

The integer sequence a stream id maps to.

Parameters

string $id

A stream id, <milliseconds>-<counter>.

Returns

int —

The sequence, monotonic in the order Redis assigned the ids.

entries()

entries() : array<string,array<string,string>>

Every entry in the stream, oldest first.

Returns

array> —

Stream id keyed to its fields.