CoolCMS API

EsEntityIndexer
in package

FinalYes

Indexes entities into ES.

Fail-open: every public method catches exceptions and only logs them (channel elasticsearch), so indexing must never bring a request down. Silently lost writes are caught up by the full reindex cron (the primary guarantee) and the incremental sync.

Tags
phpstan-type

PendingSnapshot array{ids: list, generations: array<int, string>, legacy: array<int, list>}

ReindexState array{begun: array<string, string>, afterId: int, documents: int, pending: PendingSnapshot, updatedAt: int}

Table of Contents

Constants

LOG_CHANNEL  : mixed = 'elasticsearch.index'
REINDEX_SLICE_SECONDS  : mixed = 20.0
Default time budget of one reindex slice (the scheduler gives the task about 45 s).
HEAL_LOCK_TTL  : mixed = 60
Short reindex lock under which sync repairs the write alias (held for a few requests only).
LOCK_RELEASE_LUA  : mixed = <<<'LUA' if redis.call('GET', KEYS[1]) == ARGV[...
Releases a lock by comparing the token and deleting in one step. A separate `GET` and `DEL` had a gap: the lease expired after `GET`, another run acquired the lock and the finished original owner deleted it. `KEYS[1]` = lock, `ARGV[1]` = own token.
PENDING_ACK_CHUNK  : mixed = 250
How many id and generation pairs one queue Lua script acknowledges.
PENDING_ACK_LUA  : mixed = <<<'LUA' local removed = 0 for i = 1, #ARGV, 2...
Acknowledges delivery bound to a generation: an id leaves the queue only when it still has the generation the sync or reindex read. A newer enqueue (different score) survives, atomically, so a producer cannot slip in between `ZSCORE` and `ZREM`.
PENDING_ENQUEUE_LUA  : mixed = <<<'LUA' local generation = redis.call('INCR',...
Enqueues ids into the pending queue with a new generation (`INCR` of the counter) for all ids at once. `ZADD` overwrites the score of an already queued id, so the queue always carries the generation of the latest undelivered write. `KEYS[1]` = queue sorted set, `KEYS[2]` = counter.
REINDEX_BATCH  : mixed = 200
REINDEX_LOCK_TTL  : mixed = 3600
REINDEX_SLICE_LOCK_TTL  : mixed = 300
Slice lock: short, so a lock left after a kill does not block continuation for more than a few minutes.
REINDEX_STALE_SECONDS  : mixed = 3600
A running reindex without progress for longer than this is dead and cleanup discards it.
REINDEX_STATE_TTL  : mixed = 86400
Safety net: the reindex state disappears from Redis after a day at the latest, even if cleanup did not run.
SYNC_LOCK_TTL  : mixed = 600
SYNC_OVERLAP_MINUTES  : mixed = 5
SYNC_SCAN_BATCH  : mixed = 1000

Properties

$circuitBreaker  : EsCircuitBreaker
$client  : Client
$config  : EsConfig
$ensured  : array<string, true>
$indexManager  : EsIndexManager
$redis  : Client|null
$reindexInFlight  : array<string, array{token: string, begun: array, stateKey: string, resumable: bool}>
Unfinished full reindexes (lock key mapped to owner token, created versions and state key) for cleanup after a fatal error, see {@see registerReindexShutdown()}. `resumable` means the progress is stored in Redis and the next call continues, so after a crash only the lock is released.
$reindexShutdownRegistered  : bool

Methods

__construct()  : mixed
delete()  : bool
index()  : bool
Upserts the documents of entities and deletes, by id, the documents of languages that `buildDocuments()` did not return (orphan translations after `orphanRemoval`; an empty build deletes everything, the visibility contract).
reindexAll()  : array<string, mixed>
Runs a full reindex into new index versions with an atomic alias swap (zero downtime, see `EsIndexManager`). Protected by a Redis lock with an owner token against a concurrent run.
syncSource()  : array<string, mixed>
Runs the incremental self-heal (cron every 10 min): 1) indexes entities changed since the watermark (5 min overlap), 2) orphan cleanup deletes from ES the documents of entities no longer in the DB (covers `deleteAll`, `deleteNode` and raw SQL delete), 3) moves the watermark only if everything ran without an error.
abandonReindex()  : void
Discards an unfinished reindex: the write alias goes back to the read version, the unfinished `_vN` and the Redis state are removed. The caller holds the reindex lock.
abandonStaleReindex()  : "abandoned"|"fresh"|"none"|"busy"
Cleans up a dead unfinished reindex from sync (heal path) under the reindex lock; the state is read again after the lock is acquired (another slice may have moved it meanwhile).
ackPending()  : void
Acknowledges delivery of ids from a snapshot: in the sorted set only if the id still carries the generation that was read (Lua, atomically, in chunks), in the old SET by `SREM`.
acquireLock()  : string|null
Returns the owner token when the lock is acquired, null when another run holds it.
bulk()  : int
Writes a bulk with a per-item evaluation of the response; a partial bulk failure would otherwise pass silently. Deleting a missing document (404 without `error`) is not an error.
collectEntityIds()  : array<string|int, int>
Collects all entity ids in ES (across the source bases) by `search_after` paging sorted by the entity id field, without `_source` (the id is read from the sort value).
deleteReindexState()  : void
ensureBootstrapped()  : void
Lazily bootstraps on the write path. Without it the first bulk before the first reindex cron would auto-create a concrete index named like the write alias (ES default `action.auto_create_index`) and block creating the aliases permanently. Memoized per process; the cost is about one HEAD request per source.
healWriteAliasOutsideReindex()  : void
Heals the write alias, coordinated with the reindex.
isLocked()  : bool
isReindexStale()  : bool
legacyPendingKey()  : string
Queue in the old format (a SET of ids without a generation); it is only read out and nothing new goes into it.
normalizePendingSnapshot()  : PendingSnapshot
Normalizes a queue snapshot (from Redis or the stored reindex state): the generation as an integer in a string, the old SET as id mapped to the original members (`SREM` needs the exact string).
ownsLock()  : bool
Checks the owner still holds a valid lease. Without Redis no lock exists, so true (same fail-open policy as {@see acquireLock()}); a Redis error means unknown, so false.
pendingQueueKey()  : string
pendingSequenceKey()  : string
readPending()  : PendingSnapshot
Returns a queue snapshot: ids with the generation they were read with (plus ids from the old SET).
readReindexState()  : ReindexState|null
Reads the stored progress of an unfinished reindex; null means none (or unreadable, or no Redis).
readWatermark()  : DateTimeImmutable|null
registerReindexShutdown()  : void
Cleans up after a fatal error; after a normal end of the reindex the registry is empty and it does nothing. A stored (resumable) run only releases the lock, the next call continues from the cursor; an unsaved one is rolled back (write alias back, unfinished `_vN` removed).
reindexLockKey()  : string
reindexStateAbsent()  : bool
Returns true only when the reindex state is provably missing from Redis. A Redis error means unknown, so false (healing is rather skipped). Without Redis (dev profile) there is no resumable reindex.
reindexStateKey()  : string
releaseLock()  : void
Releases the lock only for its owner: after the ttl expires, a finished reindex must not delete a foreign lock. The token comparison and `DEL` are atomic (Lua, {@see self::LOCK_RELEASE_LUA}); when `EVAL` fails the lock is not deleted blindly and expires with its ttl.
rememberPending()  : void
Queues ids whose ES write did not happen (circuit open, outage, failed bulk item).
skipWhenCircuitOpen()  : bool
stateMatchesBases()  : bool
Checks that the stored version set still matches: a base (new language) may have appeared or disappeared between slices, and the reindex then has to start again.
write()  : bool
Shared write of `index()` and `delete()`: bootstrap (ES), build operations from the source (DB/ORM), bulk (ES). A failure returns false and queues the ids as undelivered writes.
writeFailed()  : bool
writeLastReindexTimestamp()  : void
Stores the watermark of the last successful full reindex for monitoring (`EsMonitor` reports an issue when the key is missing or older than 26 h). Fail-open: a Redis error is only logged and must not fail an already finished reindex.
writeReindexState()  : bool
Returns true when the progress was stored (so the next call can continue).
writeWatermark()  : void

Constants

REINDEX_SLICE_SECONDS

Default time budget of one reindex slice (the scheduler gives the task about 45 s).

public mixed REINDEX_SLICE_SECONDS = 20.0

HEAL_LOCK_TTL

Short reindex lock under which sync repairs the write alias (held for a few requests only).

private mixed HEAL_LOCK_TTL = 60

LOCK_RELEASE_LUA

Releases a lock by comparing the token and deleting in one step. A separate `GET` and `DEL` had a gap: the lease expired after `GET`, another run acquired the lock and the finished original owner deleted it. `KEYS[1]` = lock, `ARGV[1]` = own token.

private mixed LOCK_RELEASE_LUA = <<<'LUA' if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) end return 0 LUA

PENDING_ACK_CHUNK

How many id and generation pairs one queue Lua script acknowledges.

private mixed PENDING_ACK_CHUNK = 250

PENDING_ACK_LUA

Acknowledges delivery bound to a generation: an id leaves the queue only when it still has the generation the sync or reindex read. A newer enqueue (different score) survives, atomically, so a producer cannot slip in between `ZSCORE` and `ZREM`.

private mixed PENDING_ACK_LUA = <<<'LUA' local removed = 0 for i = 1, #ARGV, 2 do local score = redis.call('ZSCORE', KEYS[1], ARGV[i]) if score and tonumber(score) == tonumber(ARGV[i + 1]) then removed = removed + redis.call('ZREM', KEYS[1], ARGV[i]) end end return removed LUA

ARGV = id, generation, id, …

PENDING_ENQUEUE_LUA

Enqueues ids into the pending queue with a new generation (`INCR` of the counter) for all ids at once. `ZADD` overwrites the score of an already queued id, so the queue always carries the generation of the latest undelivered write. `KEYS[1]` = queue sorted set, `KEYS[2]` = counter.

private mixed PENDING_ENQUEUE_LUA = <<<'LUA' local generation = redis.call('INCR', KEYS[2]) for i = 1, #ARGV do redis.call('ZADD', KEYS[1], generation, ARGV[i]) end return generation LUA

REINDEX_SLICE_LOCK_TTL

Slice lock: short, so a lock left after a kill does not block continuation for more than a few minutes.

private mixed REINDEX_SLICE_LOCK_TTL = 300

REINDEX_STALE_SECONDS

A running reindex without progress for longer than this is dead and cleanup discards it.

private mixed REINDEX_STALE_SECONDS = 3600

REINDEX_STATE_TTL

Safety net: the reindex state disappears from Redis after a day at the latest, even if cleanup did not run.

private mixed REINDEX_STATE_TTL = 86400

Properties

$ensured

private array<string, true> $ensured = []

sources with verified index and alias bootstrap (per process)

$reindexInFlight

Unfinished full reindexes (lock key mapped to owner token, created versions and state key) for cleanup after a fatal error, see {@see registerReindexShutdown()}. `resumable` means the progress is stored in Redis and the next call continues, so after a crash only the lock is released.

private array<string, array{token: string, begun: array, stateKey: string, resumable: bool}> $reindexInFlight = []

$reindexShutdownRegistered

private bool $reindexShutdownRegistered = false

Methods

index()

Upserts the documents of entities and deletes, by id, the documents of languages that `buildDocuments()` did not return (orphan translations after `orphanRemoval`; an empty build deletes everything, the visibility contract).

public index(DocumentSourceInterface $source, array<string|int, int> $ids) : bool

Returns true when the write was done (or there was nothing to do); sync moves its watermark by it.

Parameters
$source : DocumentSourceInterface
$ids : array<string|int, int>
Return values
bool

reindexAll()

Runs a full reindex into new index versions with an atomic alias swap (zero downtime, see `EsIndexManager`). Protected by a Redis lock with an owner token against a concurrent run.

public reindexAll(DocumentSourceInterface $source[, float|null $timeBudget = self::REINDEX_SLICE_SECONDS ]) : array<string, mixed>

The circuit breaker is deliberately not read: reindex is a healer and always tries ES.

The reindex runs in slices over several requests. One call processes keyset batches (id > cursor) only within $timeBudget seconds (at least one batch) and stores the progress (created _vN, cursor, document count) in Redis after each batch. The next call continues; the read alias is switched only after the last slice. A crashed slice (out of memory, timeout, kill) does not lose the progress: the shutdown function only releases the lock and the next call continues from the stored cursor. Cleanup discards a reindex without progress for over an hour (here and in syncSource()). $timeBudget = null runs the whole reindex in one request (tests, manual script).

Without Redis the progress has nowhere to go: the reindex then finishes in one request and an unfinished version is discarded after a crash.

IMPORTANT: the alias swap happens only after a fully successful fill (even a single failed bulk item means rollback); the read alias must never point to a version with holes. On failure the write alias returns to the stable version and the unfinished _vN is deleted.

Parameters
$source : DocumentSourceInterface
$timeBudget : float|null = self::REINDEX_SLICE_SECONDS
Return values
array<string, mixed> —

statistics for the cron response; key error means failure, continues: true means the slice is done and the next call finishes the rest

syncSource()

Runs the incremental self-heal (cron every 10 min): 1) indexes entities changed since the watermark (5 min overlap), 2) orphan cleanup deletes from ES the documents of entities no longer in the DB (covers `deleteAll`, `deleteNode` and raw SQL delete), 3) moves the watermark only if everything ran without an error.

public syncSource(DocumentSourceInterface $source) : array<string, mixed>

The watermark is only a latency optimization over updateDatetime/insertTime; mutations through updateColumn bypass it and are caught up only by the nightly full reindex.

Parameters
$source : DocumentSourceInterface
Return values
array<string, mixed> —

statistics for the cron response; key error means failure

abandonReindex()

Discards an unfinished reindex: the write alias goes back to the read version, the unfinished `_vN` and the Redis state are removed. The caller holds the reindex lock.

private abandonReindex(DocumentSourceInterface $source, ReindexState $state) : void
Parameters
$source : DocumentSourceInterface
$state : ReindexState

abandonStaleReindex()

Cleans up a dead unfinished reindex from sync (heal path) under the reindex lock; the state is read again after the lock is acquired (another slice may have moved it meanwhile).

private abandonStaleReindex(DocumentSourceInterface $source) : "abandoned"|"fresh"|"none"|"busy"

The recheck result is returned to the caller. When another worker started or moved a reindex (fresh) or one is running (busy), sync must not continue to write alias healing, which would point the unfinished version back to the old one.

Parameters
$source : DocumentSourceInterface
Return values
"abandoned"|"fresh"|"none"|"busy"

ackPending()

Acknowledges delivery of ids from a snapshot: in the sorted set only if the id still carries the generation that was read (Lua, atomically, in chunks), in the old SET by `SREM`.

private ackPending(DocumentSourceInterface $source, PendingSnapshot $snapshot, array<string|int, int> $ids) : void
Parameters
$source : DocumentSourceInterface
$snapshot : PendingSnapshot
$ids : array<string|int, int>

acquireLock()

Returns the owner token when the lock is acquired, null when another run holds it.

private acquireLock(string $key[, int $ttl = self::REINDEX_LOCK_TTL ][, bool $failOpen = true ]) : string|null

$failOpen: a Redis error (SET threw) means the lock is not confirmed. Reindex and sync still run without a lock then (fail-open, as without Redis), but IMPORTANT: write alias healing must not run without a confirmed lock. It switches aliases under a possibly running reindex and without the lock has no proof whether one is running.

Parameters
$key : string
$ttl : int = self::REINDEX_LOCK_TTL
$failOpen : bool = true
Return values
string|null

bulk()

Writes a bulk with a per-item evaluation of the response; a partial bulk failure would otherwise pass silently. Deleting a missing document (404 without `error`) is not an error.

private bulk(array<int, array<string, mixed>> $operations) : int

Returns the number of failed items.

Parameters
$operations : array<int, array<string, mixed>>
Return values
int

collectEntityIds()

Collects all entity ids in ES (across the source bases) by `search_after` paging sorted by the entity id field, without `_source` (the id is read from the sort value).

private collectEntityIds(DocumentSourceInterface $source) : array<string|int, int>
Parameters
$source : DocumentSourceInterface
Return values
array<string|int, int>

deleteReindexState()

private deleteReindexState(string $stateKey) : void
Parameters
$stateKey : string

ensureBootstrapped()

Lazily bootstraps on the write path. Without it the first bulk before the first reindex cron would auto-create a concrete index named like the write alias (ES default `action.auto_create_index`) and block creating the aliases permanently. Memoized per process; the cost is about one HEAD request per source.

private ensureBootstrapped(DocumentSourceInterface $source) : void
Parameters
$source : DocumentSourceInterface

healWriteAliasOutsideReindex()

Heals the write alias, coordinated with the reindex.

private healWriteAliasOutsideReindex(DocumentSourceInterface $source) : void

Runs only under the reindex lock (short ttl; a held lock means a reindex is running, so skip) and only when Redis provably holds no unfinished reindex state under that lock. The unlocked state read at the start of sync is not enough: another worker could start a reindex between it and the healing.

Parameters
$source : DocumentSourceInterface

isLocked()

private isLocked(string $key) : bool
Parameters
$key : string
Return values
bool

isReindexStale()

private isReindexStale(ReindexState $state) : bool
Parameters
$state : ReindexState
Return values
bool

normalizePendingSnapshot()

Normalizes a queue snapshot (from Redis or the stored reindex state): the generation as an integer in a string, the old SET as id mapped to the original members (`SREM` needs the exact string).

private normalizePendingSnapshot(array<string|int, mixed> $raw) : PendingSnapshot
Parameters
$raw : array<string|int, mixed>

generations: member mapped to score; legacy: list of members or id mapped to members

Return values
PendingSnapshot

ownsLock()

Checks the owner still holds a valid lease. Without Redis no lock exists, so true (same fail-open policy as {@see acquireLock()}); a Redis error means unknown, so false.

private ownsLock(string $key, string $token) : bool
Parameters
$key : string
$token : string
Return values
bool

readWatermark()

private readWatermark(string $sourceKey) : DateTimeImmutable|null
Parameters
$sourceKey : string
Return values
DateTimeImmutable|null

registerReindexShutdown()

Cleans up after a fatal error; after a normal end of the reindex the registry is empty and it does nothing. A stored (resumable) run only releases the lock, the next call continues from the cursor; an unsaved one is rolled back (write alias back, unfinished `_vN` removed).

private registerReindexShutdown() : void

Registered once per instance.

reindexStateAbsent()

Returns true only when the reindex state is provably missing from Redis. A Redis error means unknown, so false (healing is rather skipped). Without Redis (dev profile) there is no resumable reindex.

private reindexStateAbsent(DocumentSourceInterface $source) : bool
Parameters
$source : DocumentSourceInterface
Return values
bool

releaseLock()

Releases the lock only for its owner: after the ttl expires, a finished reindex must not delete a foreign lock. The token comparison and `DEL` are atomic (Lua, {@see self::LOCK_RELEASE_LUA}); when `EVAL` fails the lock is not deleted blindly and expires with its ttl.

private releaseLock(string $key, string $token) : void
Parameters
$key : string
$token : string

rememberPending()

Queues ids whose ES write did not happen (circuit open, outage, failed bulk item).

private rememberPending(DocumentSourceInterface $source, array<string|int, int> $ids) : void

IMPORTANT: without the queue "sync catches up" holds only halfway. The sync watermark sees only rows with a new insert or update time. Writes that do not bump the time (Discussion Message has no update column, admin updateColumn visibility toggles, raw SQL cascades) would be lost during an ES outage until the nightly full reindex. Sync reads the queue, reindexes the ids and removes them after success. Fail-open: a Redis error is only logged.

The queue is the sorted set pendingz:<source>, the score being the enqueue generation (counter pendingseq:<source>). A plain id SET could not tell an older replayed write from a newer undelivered one, and a success of the older sync deleted the newer write (SREM by id). The acknowledgement now deletes only the generation that was read (ackPending()). The old SET pending:<source> is only read out and acknowledged by SREM (nothing new goes into it, so SREM cannot delete a newer write); Redis drops it once empty.

Parameters
$source : DocumentSourceInterface
$ids : array<string|int, int>

stateMatchesBases()

Checks that the stored version set still matches: a base (new language) may have appeared or disappeared between slices, and the reindex then has to start again.

private stateMatchesBases(DocumentSourceInterface $source, ReindexState $state) : bool
Parameters
$source : DocumentSourceInterface
$state : ReindexState
Return values
bool

write()

Shared write of `index()` and `delete()`: bootstrap (ES), build operations from the source (DB/ORM), bulk (ES). A failure returns false and queues the ids as undelivered writes.

private write(DocumentSourceInterface $source, array<string|int, int> $ids, callable(): array<int, array<string, mixed>> $buildOperations) : bool

IMPORTANT: the breaker opens only for errors of the ES phase (bootstrap, bulk). An exception while preparing documents (fetchFresh(), buildDocuments(), id listing; a DB outage or an ORM error) is logged and the ids stay queued, but the cluster is not marked unavailable: other sources with the same prefix would lose search and writes for 60 s because of an error unrelated to ES.

Parameters
$source : DocumentSourceInterface
$ids : array<string|int, int>
$buildOperations : callable(): array<int, array<string, mixed>>
Return values
bool

writeLastReindexTimestamp()

Stores the watermark of the last successful full reindex for monitoring (`EsMonitor` reports an issue when the key is missing or older than 26 h). Fail-open: a Redis error is only logged and must not fail an already finished reindex.

private writeLastReindexTimestamp(string $sourceKey) : void
Parameters
$sourceKey : string

writeWatermark()

private writeWatermark(string $sourceKey, DateTimeImmutable $watermark) : void
Parameters
$sourceKey : string
$watermark : DateTimeImmutable
On this page

Search results