Skip to content

Commit 85c9a73

Browse files
kai-initkai-init
committed
chunk conannical publication
1 parent dba2e8e commit 85c9a73

25 files changed

Lines changed: 479 additions & 65 deletions

‎src/Cache/Engine.php‎

Lines changed: 21 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -417,9 +417,11 @@ private function readCanonicalWithResultOverlay(
417417
}
418418

419419
$overlayReason = $result->reason;
420-
$canonicalResult = $this->readCanonical(
420+
$canonicalResult = $this->readCanonicalHead(
421421
$context,
422422
$queryHash,
423+
$this->canonicalHeadFrom($head, $version),
424+
true,
423425
);
424426

425427
if ($canonicalResult->promotable()) {
@@ -506,23 +508,32 @@ private function readResultOrCanonicalProjection(
506508
return $result->withReason('result_overlay');
507509
}
508510

509-
$canonicalHead = $this->store->fetchCanonical(
510-
versionKey: $this->keys->version($context->plan->root),
511-
generationKey: $this->keys->generation($context->plan->root),
512-
tablePrefix: $this->keys->tablePrefix($context->plan->root),
513-
namespace: $context->namespace,
514-
queryHash: $canonicalQueryHash,
515-
);
516-
517511
return $this->readCanonicalProjectionFallback(
518512
$context,
519513
$queryHash,
520514
$canonicalQueryHash,
521-
$canonicalHead,
515+
$this->canonicalHeadFrom($head, $version, RedisProtocol::MEMBERSHIP),
522516
$result->reason,
523517
);
524518
}
525519

520+
/**
521+
* @param array<int, mixed> $head
522+
* @return array<int, mixed>
523+
*/
524+
private function canonicalHeadFrom(
525+
array $head,
526+
string $version,
527+
string $status = RedisProtocol::HIT,
528+
): array {
529+
$generation = RedisProtocol::resultGeneration($head);
530+
$membership = RedisProtocol::resultMembership($head);
531+
532+
return is_string($membership)
533+
? [$status, $version, $generation, $membership]
534+
: [RedisProtocol::MISS, $version, $generation];
535+
}
536+
526537
private function readCanonicalProjectionFallback(
527538
ReadContext $context,
528539
string $queryHash,

‎src/Cache/MembershipRevalidator.php‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -63,11 +63,6 @@ public function revalidate(ReadContext $context, CacheRead $read): bool
6363
}
6464

6565
$records = $this->store->mget($keys);
66-
67-
if (count($records) !== count($keys)) {
68-
return false;
69-
}
70-
7166
$guarded = array_flip($predicate);
7267

7368
foreach ($records as $payload) {

‎src/Cache/QueryEntryRepository.php‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323

2424
private const PAGINATION_LOOKAHEAD_ROWS = 1;
2525

26+
private const ESTIMATE_PROBE_MIN_ROWS = 32;
27+
2628
public function __construct(
2729
private CacheConfig $config,
2830
private CacheRuntime $runtime,
@@ -230,14 +232,30 @@ public function publishCanonical(
230232
rootVersion: $state->version,
231233
);
232234

233-
return $this->store->publishCanonical(
235+
// Rows publish in guarded slices; membership follows once they are durable.
236+
if (!$this->store->publishRows(
234237
versionKey: $this->keys->version($plan->root),
235238
generationKey: $this->keys->generation($plan->root),
236-
membershipKey: $state->key,
239+
buildingKey: $lease->buildingKey,
237240
rowKeys: $rowKeys,
238241
rowPayloads: $rowPayloads,
239242
expectedVersion: $state->version,
240243
expectedGeneration: $state->generation,
244+
rowTtl: $this->config->rowTtl,
245+
token: (string) $lease->token,
246+
leaseTtl: $this->config->buildingLockTtl,
247+
)) {
248+
return false;
249+
}
250+
251+
return $this->store->publishCanonical(
252+
versionKey: $this->keys->version($plan->root),
253+
generationKey: $this->keys->generation($plan->root),
254+
membershipKey: $state->key,
255+
rowKeys: [],
256+
rowPayloads: [],
257+
expectedVersion: $state->version,
258+
expectedGeneration: $state->generation,
241259
membershipPayload: $membership,
242260
membershipTtl: $query->configuredTtl() ?? $this->config->queryTtl,
243261
rowTtl: $this->config->rowTtl,
@@ -478,7 +496,10 @@ private function encodeResultWithinLimits(array $rows, CacheState $state): ?stri
478496
return null;
479497
}
480498

481-
if ($this->resultExceedsEstimate($rows, $state, $count)) {
499+
if (
500+
$count > self::ESTIMATE_PROBE_MIN_ROWS
501+
&& $this->resultExceedsEstimate($rows, $state, $count)
502+
) {
482503
return null;
483504
}
484505

‎src/Cache/RowRepairer.php‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
final readonly class RowRepairer
1919
{
20-
/** SQLite's traditional variable limit is 999. */
20+
// SQLite's traditional variable limit is 999.
2121
private const REPAIR_BATCH_SIZE = 900;
2222

2323
public function __construct(
@@ -159,9 +159,25 @@ private function build(
159159
$rowPayloads[] = $this->codec->encodeRow($rowsByToken[$token], $state->epoch);
160160
}
161161

162+
if (!$this->store->publishRows(
163+
versionKey: $this->keys->version($plan->root),
164+
generationKey: $this->keys->generation($plan->root),
165+
buildingKey: $lease->buildingKey,
166+
rowKeys: $rowKeys,
167+
rowPayloads: $rowPayloads,
168+
expectedVersion: $state->version,
169+
expectedGeneration: $state->generation,
170+
rowTtl: $this->config->rowTtl,
171+
token: (string) $lease->token,
172+
leaseTtl: $this->config->buildingLockTtl,
173+
)) {
174+
return null;
175+
}
176+
177+
// Waiters only wake once every repaired row is durable.
162178
if (!$this->store->publishVersionedEntries(
163-
entryKeys: $rowKeys,
164-
entryPayloads: $rowPayloads,
179+
entryKeys: [],
180+
entryPayloads: [],
165181
ttl: $this->config->rowTtl,
166182
versionKeys: [
167183
$this->keys->version($plan->root),

‎src/CacheServiceProvider.php‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,8 @@ public function boot(): void
105105
$this->app->make(Invalidator::class)->commit($name);
106106
}
107107

108+
// Registering per level keeps invalidation ahead of any afterCommit
109+
// callback the application adds at that same nesting level.
108110
try {
109111
$event->connection->afterCommit(function () use ($name): void {
110112
$this->app->make(Invalidator::class)->commit($name);

‎src/Database/QueryBuilder.php‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -337,7 +337,9 @@ protected function runSelect()
337337

338338
if ($bypass) {
339339
if ($reason !== null) {
340-
app(QueryObserver::class)->bypass($this, $reason, $statement);
340+
$observer = app(QueryObserver::class);
341+
$observer->begin();
342+
$observer->bypass($this, $reason, $statement);
341343
}
342344

343345
return $this->connection->select(
@@ -542,7 +544,9 @@ private function runBypassedExists(
542544
?string $reason,
543545
): array {
544546
if ($reason !== null) {
545-
app(QueryObserver::class)->bypass($this, $reason, $statement);
547+
$observer = app(QueryObserver::class);
548+
$observer->begin();
549+
$observer->bypass($this, $reason, $statement);
546550
}
547551

548552
return $this->connection->select(

‎src/Invalidator.php‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -215,8 +215,9 @@ public function invalidateTable(TableIdentity $table): bool
215215

216216
public function invalidateTables(array $tables): bool
217217
{
218+
// A deliberate kill switch is not a failure to report to the caller.
218219
if ($tables === [] || !$this->runtime->invalidating()) {
219-
return $tables === [];
220+
return true;
220221
}
221222

222223
$immediate = [];
@@ -388,14 +389,18 @@ private function applyMany(array $invalidations): bool
388389
try {
389390
$this->store->invalidateTableStates($states);
390391

392+
$observed = [];
393+
391394
foreach ($invalidations as $index => $invalidation) {
392-
$this->observer->invalidated(
393-
$invalidation['table'],
394-
$states[$index]['mode'],
395-
$invalidation['tokens'],
396-
);
395+
$observed[] = [
396+
'table' => $invalidation['table'],
397+
'mode' => $states[$index]['mode'],
398+
'tokens' => $invalidation['tokens'],
399+
];
397400
}
398401

402+
$this->observer->invalidatedMany($observed);
403+
399404
return true;
400405
} catch (\Throwable $exception) {
401406
$failedIndex = $exception instanceof TableInvalidationException

‎src/Lua/fetch_result_or_canonical.lua‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -6,21 +6,32 @@
66
-- ARGV[3] = canonical query hash
77
--
88
-- Returns:
9-
-- {'result', ver, payload}
9+
-- {'result', ver, payload, gen, membership?}
1010
-- {'membership', ver, gen, membership}
1111
-- {'miss', ver, gen}
1212

1313
local version = redis.call('GET', KEYS[1]) or '0'
14+
local generation = redis.call('GET', KEYS[2]) or '0'
1415
local result_key = KEYS[3] .. ':q:' .. ARGV[1] .. ':' .. ARGV[2]
15-
local result = redis.call('HGET', result_key, 'r')
16+
local membership_key = KEYS[3] .. ':q:' .. ARGV[1] .. ':' .. ARGV[3]
17+
local result, membership
1618

17-
if result then
18-
return {'result', version, result}
19+
if result_key == membership_key then
20+
local fields = redis.call('HMGET', result_key, 'r', 'm')
21+
result = fields[1]
22+
membership = fields[2]
23+
else
24+
result = redis.call('HGET', result_key, 'r')
25+
membership = redis.call('HGET', membership_key, 'm')
1926
end
2027

21-
local generation = redis.call('GET', KEYS[2]) or '0'
22-
local membership_key = KEYS[3] .. ':q:' .. ARGV[1] .. ':' .. ARGV[3]
23-
local membership = redis.call('HGET', membership_key, 'm')
28+
if result then
29+
if membership then
30+
return {'result', version, result, generation, membership}
31+
end
32+
33+
return {'result', version, result, generation}
34+
end
2435

2536
if not membership then
2637
return {'miss', version, generation}

‎src/Lua/invalidate_table.lua‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,19 @@ end
2121

2222
if mode == 'precise' then
2323
local generation = redis.call('GET', KEYS[2]) or '0'
24+
local batch = {}
2425

2526
for i = 4, #ARGV do
26-
redis.call('DEL', KEYS[3] .. generation .. ':' .. ARGV[i])
27+
batch[#batch + 1] = KEYS[3] .. generation .. ':' .. ARGV[i]
28+
29+
if #batch == 100 then
30+
redis.call('UNLINK', unpack(batch))
31+
batch = {}
32+
end
33+
end
34+
35+
if #batch > 0 then
36+
redis.call('UNLINK', unpack(batch))
2737
end
2838
end
2939

‎src/Lua/invalidate_tables.lua‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,19 @@ for key = 1, #KEYS, 4 do
2828
redis.call('INCR', KEYS[key + 1])
2929
elseif mode == 'precise' then
3030
local generation = redis.call('GET', KEYS[key + 1]) or '0'
31+
local batch = {}
3132

3233
for token = 1, token_count do
33-
redis.call('DEL', KEYS[key + 2] .. generation .. ':' .. ARGV[argument + token - 1])
34+
batch[#batch + 1] = KEYS[key + 2] .. generation .. ':' .. ARGV[argument + token - 1]
35+
36+
if #batch == 100 then
37+
redis.call('UNLINK', unpack(batch))
38+
batch = {}
39+
end
40+
end
41+
42+
if #batch > 0 then
43+
redis.call('UNLINK', unpack(batch))
3444
end
3545
end
3646

0 commit comments

Comments
 (0)