Repository navigation
Expand file tree
/
Copy pathclass-bootstrap.php
More file actions
1777 lines (1672 loc) · 73.6 KB
/
Copy pathclass-bootstrap.php
File metadata and controls
1777 lines (1672 loc) · 73.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
<?php
/**
* The substrate's WordPress boundary.
*
* The lifecycle hooks land here — activation and deactivation, `admin_init`,
* `rest_api_init`, `cron_schedules`, the minute reconcile event, the
* schedule-veto diagnostics and Site Health — and so does every question about
* the ACTIVE topology set: which workers exist, how many partitions each
* carries, which directories they read and write.
*
* @package Newspack_Nodes
*/
namespace Newspack_Nodes;
use Newspack_Nodes\Config_System\Restart_Planner;
use Newspack_Nodes\Rest\Auth_Controller;
use Newspack_Nodes\Rest\Oauth_Controller;
use Newspack_Nodes\Rest\Health_Runtime_Controller;
use Newspack_Nodes\Rest\HTTP_In_Node;
use Newspack_Nodes\Rest\SSE_Out_Node;
use Newspack_Nodes\Rest\Spawn_Controller;
\defined( 'ABSPATH' ) || exit;
/**
* WordPress-facing entry points, plus the derivations every caller makes from
* the active topology set.
*
* Every member is static because WordPress reaches all of them as callbacks —
* an activation hook, a cron action, filters — with nowhere to hold an
* instance.
*
* Wiring comes in two lazy, idempotent tiers. `ensure_diagnostics_wired()`
* touches no runtime storage, so Site Health and the cache probe still answer
* on a misconfigured base directory; `ensure_runtime_wired()` resolves that
* base and throws when it is unusable. Cron, REST and admin entry points
* report that refusal once and return rather than let it escape.
*
* @phpstan-import-type HealthResult from Health_Checks
*
* @phpstan-type Worker_Descriptor array{type: string, partition: int, topology: mixed, stale_timeout: int, on_demand_idle: int, max_runtime: int}
* @phpstan-type Resolved_Table array{kind: 'table', namespace: string, ttl: int, backend: string}|array{kind: 'row', schema: string}
*/
class Bootstrap {
/** Site Health test id (also the `test` field WP echoes back on the result). */
public const SITE_HEALTH_TEST = 'newspack_nodes_fleet';
/** The reconciliation pass's WP-Cron hook — one literal, many call sites. */
public const CRON_EVENT = 'newspack_nodes/reconcile';
/** Its recurrence, registered on `cron_schedules`. */
public const CRON_SCHEDULE = 'newspack_nodes_minute';
/** The one throttle key for a runtime base this request cannot use. */
public const RUNTIME_UNAVAILABLE = 'runtime wiring unavailable: ';
/**
* The substrate's service CIs, mount name => class, in mount order: what
* `newspack_nodes_mount_substrate_cis()` mounts on every request graph and
* what the substrate's MCP door derives its tools from.
*
* @var array<string,class-string<Service_CI_Node>>
*/
public const SUBSTRATE_CIS = [
'classes' => Rest\Classes_CI_Node::class,
'layouts' => Rest\Layouts_CI_Node::class,
'compositions' => Rest\Compositions_CI_Node::class,
'topologies' => Rest\Topologies_CI_Node::class,
'raw-logs' => Rest\Raw_Logs_CI_Node::class,
'vault' => Rest\Vault_CI_Node::class,
'aggregator' => Rest\Aggregator_CI_Node::class,
'settings' => Rest\Settings_CI_Node::class,
'status' => Rest\Status_CI_Node::class,
'sessions' => Rest\Sessions_CI_Node::class,
'workers' => Rest\Workers_CI_Node::class,
];
/**
* Health-report seam. Lazily-defaulted to a closure calling
* `Health_Checks::evaluate()`. Tests reassign it to a fixed result list so
* the Site Health rendering — status roll-up, per-result markers, badge
* color — runs as real production code rather than being mocked away.
*
* Signature: `function (): list<HealthResult>`.
*
* @var (\Closure(): list<HealthResult>)|null
*/
public static ?\Closure $health_report_evaluator = null;
/**
* `\Memcached`-construction seam. Lazily-defaulted to a closure that builds
* the real handle. Tests reassign it to an in-memory double so the
* `host:port` parsing and the empty-server-list null run as real production
* code rather than being mocked away.
*
* Signature: `function (): \Memcached`.
*
* @var (\Closure(): \Memcached)|null
*/
public static ?\Closure $memcached_factory = null;
/**
* Fleet enable/disable test seam. Production leaves this null (enabled) —
* the fleet has no production off-switch (no config field, no caller).
* Tests set false to exercise the disabled path.
*
* @var bool|null
*/
public static ?bool $fleet_enabled_override = null;
/**
* Spawn-coordinator construction seam. Lazily-defaulted to a closure
* building the real one against the configured base dir. Tests reassign to
* bind a coordinator to a temp dir without moving the whole runtime.
*
* Signature: `function (): Spawn_Coordinator`.
*
* @var (\Closure(): Spawn_Coordinator)|null
*/
public static ?\Closure $spawn_coordinator_factory = null;
/** Guards ensure_runtime_wired() so repeat entry-point calls in one request are no-ops. */
private static bool $runtime_wired = false;
/** Guards ensure_diagnostics_wired() so repeat entry-point calls in one request are no-ops. */
private static bool $diagnostics_wired = false;
/** Tracks the event entering schedule_event so a late falsy veto still has context. */
private static bool $schedule_event_context_is_reconcile = false;
/** @var array{0: array<string,list<array<array-key,mixed>>>, 1: array<string,\Throwable>}|null Request-static half of the wake map, with its failures. */
private static ?array $on_demand_wake_map = null;
/**
* Each Table `node_tables()` resolved in this process, by name, until
* `forget_node_tables()` drops them with the parsed topologies.
*
* @var array<string,array<int,array{namespace: string, ttl: int, backend: string}>>
*/
private static array $node_tables = [];
/**
* Each Row_Table `row_table()` resolved in this process, by schema, until
* `forget_node_tables()` drops them with the parsed topologies.
*
* @var array<string,array{name: string, files: array<int,string>}>
*/
private static array $row_tables = [];
/**
* Every node a gated mount built in this request — a worker's input
* Partition by `register_worker_partition()`, a Table by `mount_table()` —
* by name: the one record `HTTP_In_Node` reads before it routes a
* message that is not a command for an interpreter. `door_mounted()`
* checks the node still stands under its name.
*
* @var array<string,Node>
*/
private static array $door_mounts = [];
/** Wake-map key prefix; the active set's digest completes it and `Cache_Backend` scopes it. Rows carry offsetlog_dir. */
private const ON_DEMAND_WAKE_KEY = 'on_demand_wake_v2:';
/** Seconds a wake map survives an edited `.tsl`; a changed active set keys elsewhere. */
private const ON_DEMAND_WAKE_TTL_S = 60;
/** Row_Table key prefix; the schema word and its inputs' digest complete it. */
private const ROW_TABLE_KEY = 'row_table:';
/** Seconds a resolved Row_Table survives a change its key cannot see. */
private const ROW_TABLE_TTL_S = 300;
/**
* Substrate-version seam for `row_table()`'s cache key. Replaces the read
* of `NEWSPACK_NODES_VERSION`; tests answer another release to prove an
* upgrade misses the cache. Signature: `function (): string`.
*
* @var \Closure|null
*/
public static ?\Closure $substrate_version = null;
/**
* `readable_wake_map()`'s map, strict: an active topology whose graph will
* not read throws, after every other was read.
*
* @api Called from consumer plugins (cross-repo, invisible here): a
* newspack-nuclear-gyrobase before 1.19.0 builds every Perl render's
* spawn environment from it.
*
* @return array<string,list<array<array-key,mixed>>>
* @throws \Throwable Every unreadable topology, combined.
*/
public static function on_demand_wake_map(): array {
[ $map, $unreadable ] = self::readable_wake_map();
Worker_Should_Stop::raise( $unreadable );
return $map;
}
/**
* Resolved partition directory => the on-demand workers that TAIL it, over
* every READABLE on-demand topology, and what each unreadable on-demand
* one threw instead of answering. The answer to "did this write land
* somewhere an absent worker is waiting on", keyed by the concrete path so
* a writer needs no idea what it wrote. Which topologies read is
* `active_topologies()`'s answer; a resident one wakes nothing, so its
* failure is not for this map to report. Each caller acts on the readable
* map first and raises or reports the failures itself.
*
* Built by SUBSTITUTION, never by parsing: each Consumer source template is
* resolved through `Core::resolve_partition_template()` for that worker's
* partition — the one place `{partition}` is expanded — so a template that
* puts the token anywhere but a `.p<N>` suffix still resolves. A source
* that does not name a partition maps to the one worker `Core::owns()`
* gives it, p0, because every other worker's copy idles (ADR-33). A
* partition that is never tailed is simply absent from the map, which is
* why offsetlogs, deadletter dirs and scratch need no exclusion rule.
*
* Cached in APCu (`local_first`, memcached only as its fallback) because the
* derivation globs the user dir and every stock dir and parses every `.tsl`,
* while the askers sit on request paths. Host-LOCAL is the correct tier: the
* inputs are TSL files on disk, which differ per host. The key carries the
* active set, so activation cannot serve a stale answer. Every other input
* rides the TTL alone — an edited `.tsl`, the global `num_partitions`, the
* fleet-wide `on_demand_idle` — and a stale miss costs one cron-cadence wake.
* A map missing an unreadable topology is memoized for the request, failures
* and all, and never cached: a cached partial map would hide the failure.
*
* @api Called from consumer plugins (cross-repo, invisible here).
*
* @return array{0: array<string,list<array<array-key,mixed>>>, 1: array<string,\Throwable>} The map, then each unreadable on-demand topology's failure by name.
*/
public static function readable_wake_map(): array {
if ( null !== self::$on_demand_wake_map ) {
return self::$on_demand_wake_map;
}
// The raw CONFIG: building the key must not cost the walk it avoids.
$cache = Cache_Backend::local_first();
$key = Cache_Backend::site_key( self::ON_DEMAND_WAKE_KEY . \md5( (string) \wp_json_encode( Config::value( 'topologies' ) ) ) );
if ( null !== $cache ) {
$hit = $cache->get( $key );
if ( \is_array( $hit ) ) {
return self::$on_demand_wake_map = [ self::sanitize_wake_map( $hit ), [] ];
}
}
$entries = self::get_topologies();
[ $readable, $unreadable ] = self::split_readable( $entries );
$on_demand = \array_filter( $entries, static fn ( mixed $entry ): bool => 0 !== Worker_Base::window_of( Core::arr( $entry ), 'on_demand_idle' ) );
$map = [];
foreach ( \array_intersect_key( $readable, $on_demand ) as $topology => $entry ) {
$positions = Topology_Analyzer::consumer_positions( $topology );
foreach ( self::workers_of( $topology, $entry ) as $worker ) {
foreach ( $positions as $position ) {
if ( ! Core::owns( $position['source'], $worker['partition'] ) ) {
continue;
}
$dir = \rtrim( Core::resolve_partition_template( $position['source'], $worker['partition'], $topology ), '/' );
$cursor = '' === $position['offsetlog']
? ''
: \rtrim( Core::resolve_partition_template( $position['offsetlog'], $worker['partition'], $topology ), '/' );
// Paired so the backlog sweep never re-walks it.
$map[ $dir ][] = $worker + [ 'offsetlog_dir' => $cursor ];
}
}
}
// Only an on-demand topology's failure can hide a reader.
$unreadable = \array_intersect_key( $unreadable, $on_demand );
if ( [] === $unreadable ) {
$cache?->set( $key, $map, self::ON_DEMAND_WAKE_TTL_S );
}
return self::$on_demand_wake_map = [ $map, $unreadable ];
}
/**
* A cache entry is untrusted shape; keep only `dir => list<descriptor>`.
*
* @param array<array-key,mixed> $raw Decoded cache value.
* @return array<string,list<array<array-key,mixed>>>
*/
private static function sanitize_wake_map( array $raw ): array {
$map = [];
foreach ( $raw as $dir => $rows ) {
if ( ! \is_string( $dir ) || ! \is_array( $rows ) ) {
continue;
}
foreach ( $rows as $row ) {
if ( \is_array( $row ) ) {
$map[ $dir ][] = $row;
}
}
}
return $map;
}
/**
* Expand topologies to flat worker descriptors, one per partition (count clamped to MAX_PARTITIONS).
*
* @return array<int,Worker_Descriptor>
*/
public static function expand_workers(): array {
$workers = [];
foreach ( self::get_topologies() as $type => $entry ) {
\array_push( $workers, ...self::workers_of( $type, Core::arr( $entry ) ) );
}
return $workers;
}
/**
* One topology's worker descriptors, one per partition.
*
* @param string $type Topology name.
* @param array<array-key,mixed> $entry Its active entry.
* @return list<Worker_Descriptor>
*/
private static function workers_of( string $type, array $entry ): array {
$workers = [];
$count = self::partitions_of( $entry );
$stale = Worker_Base::window_of( $entry, 'stale_timeout' );
$idle = Worker_Base::window_of( $entry, 'on_demand_idle' );
$runtime = Worker_Base::window_of( $entry, 'max_runtime' );
for ( $p = 0; $p < $count; ++$p ) {
$workers[] = [
'type' => $type,
'partition' => $p,
'topology' => $entry['topology'] ?? '',
'stale_timeout' => $stale,
'on_demand_idle' => $idle,
'max_runtime' => $runtime,
];
}
return $workers;
}
/**
* The stale threshold one topology declares, or the default.
*
* A consumer that starts from one TYPE rather than a descriptor reads it
* here, as the render lease nuclear-gyrobase hands its Perl child does;
* `CLI` reads the same entry off its own one read of the active set for a
* whole scan. A consumer that judges staleness
* for itself falls back to the flat default, calling a worker on a topology
* that raised its threshold dead while the peer scan correctly leaves it
* running. One heartbeat, one threshold.
*
* @api Called from consumer plugins (cross-repo, invisible here).
*
* @param string $type The topology name.
* @return int Seconds.
* @throws \RuntimeException When the runtime base directory is unusable.
*/
public static function stale_timeout_for( string $type ): int {
$topologies = self::get_topologies();
if ( ! isset( $topologies[ $type ] ) ) {
return Lock_Node::STALE_TIMEOUT;
}
return Worker_Base::window_of( Core::arr( $topologies[ $type ] ), 'stale_timeout' );
}
/**
* Re-arm (init, on a cron run) the reconcile event a run that
* loaded without the substrate dropped: WordPress cannot reschedule an
* event whose schedule was never registered. The scheduled check comes
* first, so a healthy run costs one lookup in the loaded cron option.
*/
public static function rearm_reconcile_cron(): void {
if ( \wp_next_scheduled( self::CRON_EVENT ) || ! self::fleet_site() ) {
return;
}
self::self_heal_reconcile_cron();
}
/** Self-heal (admin_init): re-arm the reconcile cron if it should run but isn't scheduled. */
public static function self_heal_reconcile_cron(): void {
if ( ! self::is_fleet_enabled() ) {
return;
}
// @longform Ahead of the cron check: an install activated before the
// salt existed is healthy, so `activate()` below never runs for it and
// its cache scope stays computable. Memoized per process: one row read.
Cache_Backend::ensure_salt();
if ( ! self::runtime_base_is_available() ) {
return;
}
if ( empty( self::get_topologies() ) ) {
return;
}
if ( \wp_next_scheduled( self::CRON_EVENT ) ) {
return;
}
self::activate();
}
/**
* Schedule the reconcile cron at minute cadence: the activation hook and
* the self-heal re-arm. The `true` fifth argument asks
* `wp_schedule_event()` for a WP_Error, so a refused schedule reports its
* own code and message rather than a bare false. The wpdb Tables' shared
* tables are a wpdb Table's to install, on its first use.
*/
public static function activate(): void {
// An unsalted install has a computable cache scope.
Cache_Backend::ensure_salt();
// A row that always exists never sits in a stale notoptions list.
\add_option( Spawn_Coordinator::HOLD_OPTION, 0, '', false );
if ( ! \wp_next_scheduled( self::CRON_EVENT ) ) {
$result = \wp_schedule_event( \time() + 5, self::CRON_SCHEDULE, self::CRON_EVENT, [], true );
if ( \is_wp_error( $result ) ) {
Core::print_less_often(
'reconcile cron schedule failed: ',
\sprintf(
'code=%s message=%s schedule=' . self::CRON_SCHEDULE,
$result->get_error_code(),
$result->get_error_message()
)
);
}
}
}
/**
* Concrete dirs the Partition/Topic node `$node` writes across every ACTIVE
* topology, indexed by partition. The union: two topologies declaring the
* same node at different worker counts each contribute their own.
*
* This is how a READER finds a resource's partitions. The global
* `num_partitions` is not that number — a topology carries its own count,
* and a Topic re-partitions above it — so a reader that loops to the global
* silently sees only the low partitions.
*
* @api Called from consumer plugins (cross-repo, invisible here).
*
* @param string $node Node name as its topology declares it.
* @return array<int,string> Partition index => directory.
* @throws \Throwable When the base directory is unusable, or `$node` is not
* declared in any readable topology and an active `.tsl`
* will not read.
*/
public static function node_dirs( string $node ): array {
$dirs = [];
foreach ( self::declaring( $node, self::active_topologies() ) as $name => $entry ) {
foreach ( Topology_Analyzer::resolved_node_dirs( $name, $node, self::partitions_of( $entry ) ) as $p => $dir ) {
$dirs[ $p ] ??= $dir;
}
}
\ksort( $dirs );
return $dirs;
}
/**
* Worker indices running `$node`, across every ACTIVE topology that declares
* it. For per-partition state that never lands on disk — a memcache stats
* store keyed by the worker index — where node_dirs() has nothing to expand.
*
* @api Called from consumer plugins (cross-repo, invisible here).
*
* @param string $node Node name as its topology declares it.
* @return list<int> Partition indices, ascending.
* @throws \Throwable When the base directory is unusable, or `$node` is not
* declared in any readable topology and an active `.tsl`
* will not read.
*/
public static function node_partitions( string $node ): array {
$seen = [];
foreach ( self::declaring( $node, self::active_topologies() ) as $entry ) {
$seen += \array_fill( 0, self::partitions_of( $entry ), true );
}
$out = \array_keys( $seen );
\sort( $out );
return $out;
}
/**
* Mount every partition of each named Table into this request's graph,
* each named `Table_Node::stem()`, sinking into `_command_interpreter` and
* serving reads only. A mount lives for the rest of the request, so a
* later call mounting the same Table keeps it and builds nothing. A kept
* mount is matched by stem alone: after an activation earlier in the same
* POST changed a Table's backend or TTL, it keeps the old spec. All or
* nothing: a throw unmounts what this call built. One read of the active
* set serves every name.
*
* @api Called from consumer plugins (cross-repo, invisible here).
*
* @param list<string> $names Table names as their topologies declare them.
* @return array<string,array<int,string>> Name => partition => mounted node
* name; `[]` for a name that is not declared by any active
* topology.
* @throws \InvalidArgumentException On a name that cannot name a file.
* @throws \RuntimeException With no request graph.
* @throws Table_Unavailable When a backend cannot open, raised as thrown;
* beside a teardown that also throws, both raise
* as `Failures`, whose previous is this one.
* @throws \Throwable As node_tables().
*/
public static function mount_table( array $names ): array {
$built = [];
$ci = Core::node( Node_Names::COMMAND_INTERPRETER ) ?? throw new \RuntimeException( 'mount_table needs a request graph' );
foreach ( $names as $name ) {
try {
Table_Node::stem( $name, 0 );
} catch ( \InvalidArgumentException $e ) {
throw new \InvalidArgumentException( \esc_html( $e->getMessage() ), 0, $e );
}
}
$out = [];
try {
foreach ( self::node_tables( ...$names ) as $name => $partitions ) {
$out[ $name ] = [];
foreach ( $partitions as $p => $spec ) {
$stem = Table_Node::stem( $name, $p );
if ( ! Core::node( $stem ) instanceof Table_Node ) {
self::$door_mounts[ $stem ] = Table_Node::mount( $name, $p, $spec, $ci );
$built[] = $stem;
}
$out[ $name ][ $p ] = $stem;
}
}
} catch ( \Throwable $e ) {
throw Worker_Should_Stop::combine( [ $e, ...Worker_Should_Stop::attempt_each( $built, static fn ( string $stem ) => Core::node( $stem )?->remove_node() ) ] );
}
return $out;
}
/**
* Each named Table, partition by partition, across every ACTIVE topology
* declaring it: its namespace with its config tokens resolved, then
* `<topology>` substituted, as a worker's Shell resolves it, and
* `{partition}`, as the Table resolves it; its TTL and backend with their
* tokens resolved. Declared
* once in the `.tsl`, so the writer and every reader resolve the same
* Table. One read of the active set serves every name not yet resolved,
* and a name resolves once until `forget_node_tables()`, which every
* change to the active set, a saved `.tsl` or the config fires.
*
* @api Called from consumer plugins (cross-repo, invisible here).
*
* @param string ...$names Table names as their topologies declare them.
* @return array<string,array<int,array{namespace: string, ttl: int, backend: string}>>
* Name => partition => the Table; `[]` for a name that is not
* declared as a keyed Table by any active topology.
* @throws \Throwable When two active topologies declare one differently, a
* TTL is not a whole number of at least 1 second, a
* token that does not resolve, the base directory is
* unusable, or a name is not declared in any readable
* topology and an active `.tsl` will not read.
*/
public static function node_tables( string ...$names ): array {
$unresolved = \array_diff( $names, \array_keys( self::$node_tables ) );
if ( [] !== $unresolved ) {
$active = self::active_topologies();
foreach ( $unresolved as $name ) {
self::$node_tables[ $name ] = self::keyed( self::resolve_table( $name, $active ) );
}
}
$out = [];
foreach ( $names as $name ) {
$out[ $name ] = self::$node_tables[ $name ];
}
return $out;
}
/**
* A resolved Table's partitions as a keyed Table, its kind dropped; none
* for a Row_Table.
*
* @param array<int,Resolved_Table> $partitions resolve_table().
* @return array<int,array{namespace: string, ttl: int, backend: string}>
*/
private static function keyed( array $partitions ): array {
$out = [];
foreach ( $partitions as $p => $spec ) {
if ( 'table' === $spec['kind'] ) {
unset( $spec['kind'] );
$out[ $p ] = $spec;
}
}
return $out;
}
/**
* The one Row_Table the active topologies declare with `$schema`: its name
* and its SQLite file per partition, across every ACTIVE topology
* declaring it, each with `partitions_of()` its entry. A reader finds its
* table by schema, never by name. An active `.tsl` that will not read
* fails the lookup only when the readable topologies do not declare one
* (`may_declare()`). Whether each file exists is for the reader to find.
* One read of the active set serves the lookup, and a schema resolves
* once until `forget_node_tables()`.
*
* The answer's name and partition count are cached across processes in
* `shared_first()` for `ROW_TABLE_TTL_S`, under a key covering every
* input the resolution reads that the substrate owns: the schema, the
* substrate version, the `topologies` config, the global partition count
* and `Topology_Registry::stats()`, so a cold process parses no topology
* until one of them changes. A salt rotation moves the key with every
* other. The files are never cached, but derived from the base in force,
* and a refusal is never cached, so it is raised afresh each time.
*
* @api The row tables' read path, `Row_Table_Node::read()`.
*
* @param class-string<Row_Schema> $schema The schema the table holds.
* @return array{name: string, files: array<int,string>} Files by partition.
* @throws Table_Unavailable When the active topologies do not declare one.
* @throws \RuntimeException When the readable topologies do not declare one
* and an active `.tsl` will not read; when they
* declare several, naming each; when two
* declare its name differently; or when a
* registered topology dir will not list,
* naming it.
* @throws \InvalidArgumentException As row_tables_of().
*/
public static function row_table( string $schema ): array {
if ( isset( self::$row_tables[ $schema ] ) ) {
return self::$row_tables[ $schema ];
}
$word = self::schema_word( $schema );
$cache = Cache_Backend::shared_first();
$key = null === $cache ? '' : self::row_table_key( $schema, $word );
$hit = null === $cache ? null : $cache->get( $key );
if ( \is_array( $hit ) && \is_string( $hit['name'] ?? null ) && '' !== $hit['name']
&& \is_int( $hit['partitions'] ?? null ) && 0 < $hit['partitions']
&& Spawn_Coordinator::MAX_PARTITIONS >= $hit['partitions'] ) {
return self::$row_tables[ $schema ] = self::row_table_files( $hit['name'], $hit['partitions'] );
}
return self::resolved_row_table( $schema, $word, $cache, $key );
}
/**
* `row_table()` resolved from the active topologies, never from the
* shared cache or the process memo, for a path that writes or deletes the
* table's files and so must not act on an answer up to `ROW_TABLE_TTL_S`
* old. What it resolves replaces both, so a read after it agrees.
*
* @api The row tables' write paths: `Row_Table_Node::build()`, and
* newspack-intelligence's `ontology reproject`, cross-repo.
*
* @param class-string<Row_Schema> $schema The schema the table holds.
* @return array{name: string, files: array<int,string>} Files by partition.
* @throws Table_Unavailable As `row_table()`.
* @throws \RuntimeException As `row_table()`.
* @throws \InvalidArgumentException As `row_table()`.
*/
public static function resolve_row_table( string $schema ): array {
$word = self::schema_word( $schema );
$cache = Cache_Backend::shared_first();
return self::resolved_row_table( $schema, $word, $cache, null === $cache ? '' : self::row_table_key( $schema, $word ) );
}
/**
* `row_table()`'s shared-cache key, built from what the resolution reads
* without reading a topology: the user dir it covers is registered first.
*
* @param string $schema The schema class.
* @param string $word Its schema word.
*/
private static function row_table_key( string $schema, string $word ): string {
self::ensure_runtime_wired();
$version = self::$substrate_version ?? static fn (): string => NEWSPACK_NODES_VERSION;
$inputs = [ $schema, $version(), Config::value( 'topologies' ), self::global_num_partitions(), Topology_Registry::stats() ];
return Cache_Backend::site_key( self::ROW_TABLE_KEY . $word . ':' . \md5( (string) \wp_json_encode( $inputs ) ) );
}
/**
* Resolve `$schema`'s Row_Table from the active topologies, then store it
* in the process memo and, under `$key`, the shared cache.
*
* @param class-string<Row_Schema> $schema The schema the table holds.
* @param string $word Its schema word.
* @param Cache_Backend|null $cache The shared cache, or null.
* @param string $key `row_table_key()`, or '' with no cache.
* @return array{name: string, files: array<int,string>}
* @throws Table_Unavailable When the active topologies do not declare one.
* @throws \RuntimeException As `row_table()`.
* @throws \InvalidArgumentException As row_tables_of().
*/
private static function resolved_row_table( string $schema, string $word, ?Cache_Backend $cache, string $key ): array {
$active = self::active_topologies();
[ $readable, $unreadable ] = $active;
$names = self::row_tables_of( $schema, ...\array_map( 'strval', \array_keys( $readable ) ) );
if ( [] === $names ) {
Worker_Should_Stop::raise( self::may_declare( $unreadable ) );
throw new Table_Unavailable( \esc_html( "no active topology declares a Row_Table of schema {$word}" ) );
}
if ( 1 < \count( $names ) ) {
throw new \RuntimeException( \esc_html( 'the active topologies declare ' . \count( $names ) . " Row_Tables of schema {$word}: " . \implode( ', ', $names ) ) );
}
$partitions = \count( self::resolve_table( $names[0], $active ) );
$cache?->set( $key, [ 'name' => $names[0], 'partitions' => $partitions ], self::ROW_TABLE_TTL_S );
return self::$row_tables[ $schema ] = self::row_table_files( $names[0], $partitions );
}
/**
* A Row_Table's name and its file per partition.
*
* @param string $name Table name.
* @param int $partitions Partition count.
* @return array{name: string, files: array<int,string>}
*/
private static function row_table_files( string $name, int $partitions ): array {
$files = [];
for ( $p = 0; $p < $partitions; ++$p ) {
$files[ $p ] = Table_Node::file( $name, $p );
}
return [
'name' => $name,
'files' => $files,
];
}
/**
* Every Row_Table the named topologies declare with `$schema`, each once,
* in the order they declare them, whether or not the topologies run.
*
* @api Uninstall's read of a plugin's topologies, cross-repo.
*
* @param class-string<Row_Schema> $schema The schema the tables hold.
* @param string ...$topologies Topology names.
* @return list<string>
* @throws \InvalidArgumentException When `$schema` does not name a
* `*_Row_Schema`.
* @throws \RuntimeException As `Topology_Analyzer::declared_tables()`.
*/
public static function row_tables_of( string $schema, string ...$topologies ): array {
$word = self::schema_word( $schema );
$names = [];
foreach ( $topologies as $topology ) {
foreach ( Topology_Analyzer::declared_tables( $topology ) as $name => $declared ) {
if ( 'row' === $declared['kind'] && $word === $declared['schema']
&& $schema === Command_Interpreter_Node::resolve_named( $word, '_Row_Schema', Row_Schema::class ) ) {
$names[ $name ] = true;
}
}
}
return \array_map( 'strval', \array_keys( $names ) );
}
/**
* The word a `.tsl` names `$schema` by: its short name less `_Row_Schema`.
*
* @param string $schema A Row_Schema class.
* @throws \InvalidArgumentException When it does not name a `*_Row_Schema`.
*/
private static function schema_word( string $schema ): string {
$short = \substr( \strrchr( '\\' . $schema, '\\' ) ?: '', 1 );
if ( ! \str_ends_with( $short, '_Row_Schema' ) || ! \is_a( $schema, Row_Schema::class, true ) ) {
throw new \InvalidArgumentException( \esc_html( "{$schema} does not name a Row_Schema" ) );
}
return \substr( $short, 0, -\strlen( '_Row_Schema' ) );
}
/**
* Which active topologies READ, answered once for every caller that reads
* a graph: each readable name with its `get_topologies()` entry, and what
* each other active name threw instead, keyed by name. A name with no
* `.tsl` fails naming itself rather than dropping out in silence.
*
* Each consumer acts on the readable set and surfaces the failures its own
* way — a row per topology in a listing, a raise after the work is done —
* so one broken `.tsl` costs its own rows and nothing else. Readable means
* both the graph and the write set build, the latter reading every owned
* Table off its owner's line; the graph walk memoizes success and failure
* alike, and the write set its success, so asking again costs no parse.
*
* @return array{0: array<string,array<array-key,mixed>>, 1: array<string,\Throwable>} Readable name => entry, then name => what it threw, in configured order.
* @throws \RuntimeException When the runtime base directory is unusable.
*/
public static function active_topologies(): array {
return self::split_readable( self::get_topologies() );
}
/**
* Active topology set: the `newspack_nodes/topologies` catalog narrowed to
* the names the `topologies` config key selects.
*
* Anything but a list of names — the empty default included — activates
* nothing, so an install spawns no workers until an operator opts one in. A
* selected name the catalog does not carry is synthesized from its `.tsl`
* frontmatter, and left out only when the name does not resolve to any
* `.tsl`, so a topology registered after the catalog is published still
* spawns.
*
* This is the CONFIGURED set: what spawns, what holds lock dirs, how many
* partitions each carries. None of that reads a graph, so an entry whose
* graph will not read is here too. A caller that reads graphs takes
* `active_topologies()`, which also names every name left out here.
*
* @return array<string,mixed> Topology name => entry (keys are always non-empty strings).
*/
public static function get_topologies(): array {
$catalog = self::get_topology_catalog();
$default_np = self::global_num_partitions();
$default_idle = self::config_on_demand_idle();
$active = [];
foreach ( self::active_names() as $name ) {
$entry = $catalog[ $name ] ?? Topology_Registry::synthesize_entry( $name, $default_np, Lock_Node::STALE_TIMEOUT, $default_idle );
if ( null !== $entry ) {
$active[ $name ] = $entry;
}
}
return $active;
}
/**
* `active_topologies()` over entries the caller already built, so a caller
* that needs the configured entries too walks the catalog once.
*
* @param array<string,mixed> $entries `get_topologies()`.
* @return array{0: array<string,array<array-key,mixed>>, 1: array<string,\Throwable>}
*/
public static function split_readable( array $entries ): array {
$names = self::active_names();
$failures = Worker_Should_Stop::attempt_each(
\array_combine( $names, $names ),
static function ( string $name ) use ( $entries ): void {
if ( ! isset( $entries[ $name ] ) ) {
throw new Topology_Missing( \esc_html( "unknown topology '$name': it does not resolve to a .tsl file" ) );
}
Topology_Analyzer::graph_for( $name );
// Only the write set reads each owner's line, not the graph.
Topology_Analyzer::write_set( $name );
}
);
return [ \array_map( Core::arr( ... ), \array_diff_key( $entries, $failures ) ), $failures ];
}
/**
* The Tables `$names` resolve to across the given readable topologies
* alone, uncached, by kind: the answer for topologies outside the active
* set, whose Tables `node_tables()` does not see.
*
* @param array<string,array<array-key,mixed>> $readable Topology name => entry.
* @param string ...$names Table names as their topologies declare them.
* @return array<string,array<int,Resolved_Table>> Name => partition => the Table; `[]` for an undeclared name.
* @throws \Throwable When those topologies declare one differently, or a TTL is not a whole number of at least 1 second.
*/
public static function tables_of( array $readable, string ...$names ): array {
$out = [];
foreach ( $names as $name ) {
$out[ $name ] = self::resolve_table( $name, [ $readable, [] ] );
}
return $out;
}
/**
* One Table resolved across `$active`, partition by partition: a keyed
* Table's tokens resolved and its namespace per partition, or a
* Row_Table's schema word. Declarations that resolve alike union their
* partitions, the way node_dirs() does, however each is spelled; one
* resolving differently, a keyed Table beside a Row_Table included, is a
* second Table under one name.
*
* @param string $name Table name.
* @param array{0: array<string,array<array-key,mixed>>, 1: array<string,\Throwable>} $active active_topologies().
* @return array<int,Resolved_Table> Partition => the Table.
* @throws \Throwable As node_tables().
*/
private static function resolve_table( string $name, array $active ): array {
$first = null;
$partitions = 0;
foreach ( self::declaring( $name, $active ) as $topology => $entry ) {
$declared = Topology_Analyzer::declared_tables( $topology )[ $name ] ?? null;
if ( null === $declared ) {
continue;
}
$spec = 'row' === $declared['kind'] ? $declared : [
'kind' => 'table',
'namespace' => \str_replace( [ '<topology>', '{topology}' ], $topology, Core::resolve_config_tokens( $declared['namespace'], true ) ),
'ttl' => Core::canonical_decimal( Core::resolve_config_tokens( $declared['ttl'], true ), false )
?? throw new \RuntimeException( \esc_html( "Table {$name} declares a TTL that is not a whole number of at least 1 second: {$declared['ttl']}" ) ),
'backend' => Core::resolve_config_tokens( $declared['backend'], true ),
];
if ( null !== $first && $first[0] !== $spec ) {
throw new \RuntimeException( \esc_html( "Table {$name} is declared differently by {$first[1]} and {$topology}" ) );
}
$first ??= [ $spec, $topology ];
$partitions = \max( $partitions, self::partitions_of( $entry ) );
}
$out = [];
for ( $p = 0; null !== $first && $p < $partitions; ++$p ) {
$out[ $p ] = 'table' === $first[0]['kind'] ? \array_replace( $first[0], [ 'namespace' => Core::resolve_partition_template( $first[0]['namespace'], $p ) ] ) : $first[0];
}
return $out;
}
/**
* The readable active topologies declaring `$node`, name => entry. One
* that will not read costs only its own share of the answer; when the
* readable ones do not declare the node, every `.tsl` that will not read
* raises (`may_declare()`), because the node may be exactly what it
* declares.
*
* @param string $node Node name as its topology declares it.
* @param array{0: array<string,array<array-key,mixed>>, 1: array<string,\Throwable>} $active active_topologies(), read once by the caller.
* @return array<string,array<array-key,mixed>>
* @throws \Throwable Every active `.tsl` that will not read, combined, when
* the readable ones do not declare `$node`.
*/
private static function declaring( string $node, array $active ): array {
[ $readable, $unreadable ] = $active;
$declaring = \array_filter(
$readable,
static fn ( string $name ): bool => Topology_Analyzer::declares_node( $name, $node ),
\ARRAY_FILTER_USE_KEY
);
if ( [] === $declaring ) {
Worker_Should_Stop::raise( self::may_declare( $unreadable ) );
}
return $declaring;
}
/**
* The unreadable topologies a lookup answers for: each whose `.tsl` will
* not read, which may declare what is asked. A name that does not
* resolve to any `.tsl` (`Topology_Missing`) declares nothing, so a lookup
* never raises it.
*
* @param array<string,\Throwable> $unreadable What each unreadable active name threw.
* @return array<string,\Throwable>
*/
private static function may_declare( array $unreadable ): array {
return \array_filter( $unreadable, static fn ( \Throwable $e ): bool => ! $e instanceof Topology_Missing );
}
/**
* Whether the `topologies` config key selects `$name`: the configured set,
* read without walking the catalog.
*
* @param string $name Topology name.
*/
public static function is_active( string $name ): bool {
return \in_array( $name, self::active_names(), true );
}
/**
* The names the `topologies` config key selects, deduplicated, in order,
* whether or not a `.tsl` resolves for each: the answer reads no graph.
*
* @return list<non-empty-string>
*/
public static function active_names(): array {
$names = Config::value( 'topologies' );
return \array_values( \array_unique( \array_filter(
\is_array( $names ) ? $names : [],
static fn ( mixed $name ): bool => \is_string( $name ) && '' !== $name
) ) );
}
/**
* A Vault mutation re-credentials the spokes, so every worker holding a
* vault-consuming node must RE-READ them rather than serve stale credentials
* for the rest of its ~10-minute lifetime. The reload channel, never the
* restart one: a credential change must not cost a process recycle. A
* `Vault_Group` also reconciles its members on this same RELOAD, building
* new children and retracting departed ones.
*
* Which topologies those are is DERIVED from each active topology's parsed
* graph, never a topology name — names are deployment config (renamable,
* user-dir-shadowable) and a name-keyed signal drifts silently into a no-op.
* The consumers are named as classes, matched by ancestry, so every broker
* counts through its base `Remote_Broker_Node` and a new one needs no
* entry here. `Vault_Group` is its own entry rather than relying on its
* children's class: a group whose last member just left derives no
* broker at all, and its own type is what still counts it as a consumer.
* The reset below does not decide that match — a cold read already sees
* the current Vault — it exists so a reader LATER in this same process
* (the flattened statements, the graph, the write set) sees the Vault as
* just written rather than whatever an earlier read in this process
* cached.
*
* `Restart_Planner::request_reloads()` signals every readable consumer
* whatever another topology or lock dir refused, then raises the refusals,
* so the Vault write that fired this action reports them.
*
* @throws \Throwable An unusable lock tree; every unreadable topology and refused signal, after the rest were signalled.
*/
public static function reload_vault_consumers(): void {
if ( ! self::fleet_site() ) {
return; // Subsite: the fleet is network-global and runs on the main site.
}
Topology_Analyzer::reset_caches();
Restart_Planner::request_reloads( Config::get_base_directory_with_locks(), [ Remote_Broker_Node::class, Vault_Group_Node::class ] );
}
/**
* Canonical partition count for a topology, as the fleet SPAWNS it: a name
* the catalog carries takes `partitions_of()` of that entry, exactly as
* `expand_workers()` does, and only a name the catalog lacks is synthesized
* from its TSL frontmatter. It serves names that may be inactive, since the
* admin localizer and the `topologies.dump` command list every registered
* topology; a caller holding an active entry reads `partitions_of()`.
*
* A caller looping names passes the catalog in, built once for the loop.
*
* @param string $name Topology name.
* @param array<array-key,mixed>|null $catalog `get_topology_catalog()`, when the caller already built it.
* @return int Partition count in [1, MAX_PARTITIONS].
*/
public static function num_partitions_for( string $name, ?array $catalog = null ): int {
$catalog ??= self::get_topology_catalog();