frust-engine 0.6.2

Sparse-strip GPU render pipeline for Frust: compiles a scene display list into strips and records the draw passes on wgpu.
Documentation
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
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
//! Scheduling a recording into render passes.
//!
//! A recorded frame is a tree: a root command stream that may enter a layer,
//! whose own stream may enter another. Draws inside a layer that composites
//! with anything other than plain source-over at full opacity cannot be written
//! straight to the surface — the layer has to be rendered in isolation first
//! and then composited as a whole. A GPU render pass writes one target, so each
//! isolated layer costs a pass of its own. This module decides how many passes
//! a frame needs, what each one targets, and in what order they run; it
//! allocates nothing and touches no device.
//!
//! [`Schedule::build`] returns the frame's [`Round`]s in execution order. A
//! round is one render pass over either the frame's own surface or one pooled
//! intermediate [page](pages). Its [`ops`](Round::ops) run in the order they
//! are listed: a batch of draws, then the composite of a finished page onto the
//! round's target, then more draws, and so on, exactly as the recording
//! interleaved them.
//!
//! One target can take several rounds. The surface, and a layer's page, are
//! written by as many passes as the frame needs; each after the first loads what
//! the one before it left rather than clearing (see [`PageTarget::continued`]),
//! so the sequence reads as one painter's-order walk however it was cut up.
//!
//! The surface's clear and the frame's opaque strips are not passes of their
//! own: the renderer draws them inside the pass of the first surface round that
//! draws, which clears the surface instead of loading it (see
//! [`crate::renderer`]'s *The frame's passes*), so a frame with no isolated
//! layer and no hole punch is one render pass in all.
//!
//! ## What this scheduler serves, and what it refuses
//!
//! Frust's layer needs are narrow. Clips lower to a scissor rectangle or a
//! coverage mask and never to an intermediate texture (see
//! [`compile::clip`](crate::compile::clip)), so the overwhelming majority of
//! frames contain no layer at all and schedule to a single round. Opacity
//! layers exist but are rare and shallow.
//!
//! The scheduler serves every layer tree [`MAX_LIVE_PAGES`] pooled pages are
//! enough for:
//!
//! - isolated layers nested at most [`MAX_CHAIN_DEPTH`] deep;
//! - isolated layers *beside* each other under one parent, a fan of any width —
//!   two widgets fading at once is the commonest sibling shape a frust screen
//!   records, and a staggered list entrance fading five rows at once is served
//!   on the same two pages;
//! - a chain hanging off an isolated ancestor's later child, which is what a
//!   navigation transition records for the whole of a scrub: a full-screen page
//!   layer holding a chip that carries a translucent chip of its own. This is
//!   the shape the [spill page](PageParity::Spill) exists for — see *The spill
//!   page* below;
//! - any mixture of the above that stays inside [`MAX_LIVE_PAGES`] live pages;
//! - a Gaussian-blur or shadow-only drop-shadow [filter](crate::filters) layer
//!   recorded directly under the frame's own surface, which takes *both*
//!   ping-pong groups for the length of its pass sequence (see *Filter rounds*
//!   below).
//!
//! A fan of any width fits because a parent does not have to composite *all* of
//! its isolated children in one pass. When both groups are taken, the parent's
//! round is [cut](cut_at): the ops accumulated so far are emitted as a round of
//! their own, the pages that batch composites go back to the pool, and the next
//! sibling is rendered into one of them. What a cut cannot pay for is a layer
//! whose own round samples a live page while an isolated ancestor is holding
//! another — three pages really are live at that point — and that is where the
//! spill page comes in.
//!
//! Everything else — a layer that finds no page even after a cut and the spill,
//! a deeper chain, a filter this engine does not render, a filter layer nested
//! inside another recorded layer, a non-default blend mode, a layer mask or
//! layer clip path — is refused with [`EngineError::SchedulerEscalation`]
//! carrying a reason that names what was found.
//!
//! ## A refusal is a skipped frame, not a fallback
//!
//! Refusing is a return value and never a panic (E17), but it is not a route to
//! a second renderer: the engine tier carries none, and which renderer draws a
//! surface is decided once when the surface is configured, not per frame. What
//! becomes of a refused frame is the caller's own contract. `frust-render`'s
//! engine arm drops it — the acquired texture is released unpresented, the
//! refusal is counted, and the reason is logged rate-limited — so nothing new
//! reaches the screen and whatever was presented last stays there.
//!
//! A shape that refuses on *every* frame therefore freezes a surface rather
//! than degrading it, which is the reason the refused set is kept to shapes
//! frust's own widget tree does not record, and the reason a shape it does
//! record (two fading siblings) is served here instead of escalated. Porting the
//! reference renderer's general scheduler, which serves any scene at the cost of
//! roughly ten times this module's code, is what the
//! [`full-scheduler`](full) feature marks the site for.
//!
//! ## Bottom-up traversal and the two-page ping-pong
//!
//! Rounds are emitted contents-before-composite: a layer's contents must exist
//! before the pass that composites them can run, so the deepest layer is
//! rendered first and the surface's last round is last. A layer takes the group
//! its depth's parity names when that group is free, and the other group when it
//! is not (see [`PageParity`]).
//!
//! Parity alone is what bounds a *chain* to two live pages however deep it runs:
//! while a parent renders into its own page it samples the child's, and the
//! child's page returns to the pool the moment the parent's pass ends, so the
//! grandchild reuses it. Siblings are what makes the fallback to the other group
//! necessary — two layers at one depth share a parity, and a second page in one
//! group would overwrite the first before the parent ever sampled it. So the
//! second sibling takes the group the first left free, and a third finds both
//! taken. That is what [`cut_at`] answers: the parent's round is closed early,
//! its two composites are done, both pages come back, and the third sibling
//! takes the group the first one had. A fan of any width costs one extra pass
//! per pair rather than a refused frame.
//!
//! Pages are named lazily, by the round that renders into them, rather than
//! reserved on the way down. An outer layer that reserved its page before the
//! descent would hold one group for the whole traversal and defeat the
//! ping-pong. The one exception is a layer whose round has to be cut: a cut
//! renders into the layer's own page, so that layer takes its group at its first
//! cut and keeps it until its last round. That is what makes a cut ancestor
//! expensive for everything below it — from its first cut onwards it is holding
//! one of the two groups, so a chain hanging off its later child has only one
//! group left to ping-pong in.
//!
//! ## The spill page
//!
//! One page beside the pair ([`PageParity::Spill`]), taken only where the walk
//! would otherwise refuse the frame: after [`make_room`] has cut everything it
//! could and both groups are still holding pages later rounds composite, a
//! *regular* layer takes the spill page instead of escalating. It is acquired
//! and released exactly as a parity page is — it goes back to the pool the
//! moment the round whose composite sampled it ends — so it is one more live
//! intermediate at a peak, not a page held for the frame.
//!
//! It exists for one shape, and that shape is a navigation transition: a
//! full-screen layer at a fractional opacity (the page being scrubbed) holding
//! a chip beside a chip that carries a translucent chip of its own. The outer
//! layer takes a group at the cut its first chip forces, the nested chip takes
//! the other, and the chip hosting it — whose own round has to sample the
//! nested page while the outer page is still owed upwards — has none. Because
//! the opacity is fractional for the whole gesture, refusing it refuses *every*
//! frame of the gesture: the surface holds its last presented image until the
//! transition ends, which is a frozen scrub rather than a dropped frame (see *A
//! refusal is a skipped frame* above). The identical content at the root
//! schedules on two pages, which is what made the shape a transition-only
//! defect.
//!
//! Bounded at one, deliberately. A second layer wanting the spill while it is
//! held is refused exactly as before, which is what keeps a frame's live
//! intermediates at [`MAX_LIVE_PAGES`] and keeps this a narrowing of the
//! refused set rather than a step towards the general scheduler. Two callers do
//! not reach for it at all: a [cut](cut_at) never takes it (a cut that cannot
//! find a group is skipped, and the walk carries on to the acquire that can
//! spill), and neither does a [filter layer](filter_rounds), whose passes
//! ping-pong between two groups of their own and so are held to the pair.
//!
//! Everything the pair already served schedules exactly as it did:
//! [`LivePages::free`] counts only the two groups, so the cutting decisions
//! that serve fans and chains are made against the state they were tuned
//! against, and the spill is reached only at the site that used to escalate.
//!
//! ## Inlined and dropped layers
//!
//! A layer only needs isolating when compositing it differs from drawing its
//! contents directly. A recorded layer at full opacity with no blend, mask or
//! clip composites source-over at alpha 1, which is precisely what drawing its
//! contents into the parent does, so its stream is spliced into the parent's
//! round and it costs no page and no pass. At the other end, a layer at zero
//! opacity contributes nothing at all and is dropped along with everything
//! nested inside it, and so does a layer whose contents cover no pixel at all —
//! neither its own round nor any round its descendants would have rendered is
//! emitted, because nothing would ever sample the pages they wrote.
//!
//! Depth is therefore counted over *isolated* layers only. Counting recorded
//! depth instead would let an inlined layer push its isolated descendant onto
//! the same parity as an isolated ancestor, putting two live pages in one group.
//!
//! ## Filter rounds
//!
//! A [filter](crate::filters) layer is an isolated layer whose page is not
//! composited straight away: a sequence of filter passes runs over it first,
//! and the parent composites what the last of them wrote. A pass reads a whole
//! image and writes a whole image, and one render pass cannot do both to one
//! texture, so each pass is a round of its own writing the group it did not
//! read — the same ping-pong the chain uses, run between two pages of one
//! layer instead of two layers. The sequence is arranged to be even in length
//! (see [`crate::filters::blur::blur_passes`] and
//! [`crate::filters::drop_shadow::drop_shadow_passes`]), so the result lands
//! back in the page the layer's own contents were rendered into and the
//! composite is the ordinary one.
//!
//! Every filter round clears its destination rather than loading it. A pass
//! writes only the region its step names — a decimated one writes a quarter of
//! the texels the pass before it did — and the kernels sample bilinearly with
//! no bounds checks, so whatever surrounds the written region has to be
//! transparent rather than a previous holder's pixels.
//!
//! Two consequences bound what filter shapes are served. The layer holds *both*
//! groups from its first filter round to its last, so a filter layer is refused
//! whenever the second group cannot be handed back to it; and it is refused
//! outright when it is recorded inside another layer, because `vello_common`
//! places a filter layer in its parent by undoing a source shift the reference
//! renderer applies to a filter layer's contents and frust's compiler does not,
//! which would size the parent's page from bounds short of the filter's own
//! spread on two sides. Neither is a shape frust records: a backdrop blur is a
//! layer under the surface.
//!
//! ## Retained pages
//!
//! A layer the renderer's retained-page store admitted
//! ([`crate::cache::retained`]) is handed to [`Schedule::build_retained`] with
//! a [`RetainedRole`]. A *recorded* body is walked like an isolated layer — its
//! own nested translucent layers take pooled pages exactly as they would inside
//! any layer — but its rounds target the store's page
//! ([`RoundTarget::Retained`]) rather than a pooled one, it is never inlined at
//! full opacity or dropped at zero (the page is wanted for later frames
//! whatever this one composites), and its composite samples that page
//! ([`Composite::retained`]). A *drawn* one is an empty layer standing at its
//! painter-order position, which becomes the composite alone. Neither takes a
//! ping-pong group or the spill page, and neither counts toward
//! [`MAX_LIVE_PAGES`] or [`MAX_CHAIN_DEPTH`]: the store's page lives outside the
//! pool for the store's own lifetime, so nothing about it competes for the
//! frame's live intermediates. A retained round is cut like an isolated
//! layer's when an ancestor needs room, and continues its own page after the
//! cut.
//!
//! ## Backdrop rounds
//!
//! A backdrop blur ([`crate::filters`]' *The backdrop blur*) is handed to
//! [`Schedule::build_frame`] as a recorded layer holding one draw. Where the
//! walk is drawing into the frame's own target — the root, or layers inlined
//! into it — it costs, in order: a **cut** of the surface round open so far
//! (unconditional, so every op recorded before the backdrop lands before
//! anything reads the target back, including the composite of a retained page
//! drawn ahead of it), a **region copy** round ([`RegionCopy`]) seeding a page
//! with the backdrop's region of the target, the blur's **pass rounds** over
//! that page exactly as a filter layer's run, and then the backdrop's draw
//! spliced into the root's next surface round, which samples the blurred page
//! ([`BackdropSample`]) and releases it when it ends. The page is held across
//! later ops like a pending composite is: a page-hungry layer recorded after
//! the backdrop cuts that surface round to get it back.
//!
//! Anywhere else — inside an isolated layer, a retained body, or a layer
//! covering nothing — there is no frame beneath the backdrop to sample, so its
//! layer is dropped: no round, and no backdrop draw in any page round. A frame
//! carrying no backdrop schedules exactly as [`Schedule::build_retained`] does.

pub mod pages;

pub use pages::{PageConfig, PageParity, PageSize, filter_page_size, page_ceiling, page_size};

use core::ops::Range;

use frust_gpu::TierCaps;
use vello_common::geometry::{RectU16, SizeU16};
use vello_common::peniko::BlendMode;
use vello_common::record::{CommandRecorder, Node, RecordedLayer, RecordedLayerKind};

use crate::compile::EngineDraw;
use crate::error::EngineError;
use crate::filters::{Backdrop, FilterStep, ServedFilter, blur, drop_shadow, served_filter};

/// The deepest chain of nested isolated layers this scheduler serves.
///
/// Four levels covers every layer shape frust's widget set records; a deeper
/// one is escalated rather than served — the frame is skipped, and shapes past
/// this bound are rare enough that skipping them beats the cost and complexity
/// of growing this module to serve them.
pub const MAX_CHAIN_DEPTH: usize = 4;

/// The page groups the depth-parity ping-pong alternates between.
///
/// Two is what the even/odd alternation guarantees for a chain of any depth,
/// and what a fan of any width is served on by cutting its parent's round. It
/// is the *pair*, not the frame's page budget — see [`MAX_LIVE_PAGES`].
pub const PING_PONG_GROUPS: usize = 2;

/// The most intermediate pages this scheduler keeps live at one time.
///
/// The [two ping-pong groups](PING_PONG_GROUPS) plus one: a single spill page
/// ([`PageParity::Spill`]) for the shape that needs a third live page and
/// cannot be cut into needing fewer. A shape needing a *fourth* — after cutting
/// an open round has been tried and could not hand one back — is a shape this
/// scheduler does not serve.
pub const MAX_LIVE_PAGES: usize = PING_PONG_GROUPS + 1;

/// Where the reference renderer's general scheduler is ported.
///
/// Deliberately empty. The scheduler this module implements covers the layer
/// shapes frust actually records and escalates the rest; the general algorithm
/// — an atlas per parity group, batching independent layers into shared rounds,
/// filter pass plans, non-default blend through a scratch texture — is an order
/// of magnitude more code and lands here, behind this feature, if and when a
/// frame shape needs it. The feature exists now so the module path is settled
/// and the escalation route above has a named destination.
#[cfg(feature = "full-scheduler")]
pub mod full {}

/// One pooled intermediate page a round renders into.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PageTarget {
    /// The layer this page holds, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    pub layer: u32,
    /// The layer's depth counted over isolated layers only, one-based.
    pub depth: usize,
    /// The group the page comes from — [`depth`](Self::depth)'s parity when
    /// either of the pair was free, the other of the pair when it was not, and
    /// the [spill page](PageParity::Spill) when neither was.
    pub parity: PageParity,
    /// The extent to acquire the page at.
    pub size: PageSize,
    /// The tile-aligned device-space bounds this page holds — the layer's own
    /// for a layer that fits one page, and one column [band](pages::page_bands)
    /// of them for a layer wider than any page is.
    ///
    /// The contents are rendered at the page's origin, so every strip drawn
    /// into this round is offset by `-(bounds.x0, bounds.y0)` *and clipped to
    /// this rectangle*: a banded layer's rounds are all handed the same ops, so
    /// this is what selects the part of them each band actually holds. The
    /// renderer's `PageWindow` is where both halves are applied.
    pub bounds: RectU16,
    /// Whether an earlier round of this same layer already rendered into this
    /// page, so this one loads its contents instead of clearing them.
    ///
    /// A pooled page holds whatever its last holder left there, which is why a
    /// layer's *first* round always clears it. A layer whose round was
    /// [cut](cut_at) keeps the page across the cut — clearing again would wipe
    /// the half already drawn — and no other layer can have taken the group in
    /// between, because the cut is what made this layer the group's holder.
    pub continued: bool,
}

/// What a round renders into.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RoundTarget {
    /// The frame's own surface. Draws land at their recorded coordinates.
    Root,
    /// A pooled intermediate page holding one isolated layer.
    Page(PageTarget),
    /// A retained page holding one recorded body — the store's own texture,
    /// never a pooled one (see the module header's *Retained pages*).
    Retained(RetainedTarget),
}

/// Compositing a finished page onto the round's target.
///
/// One instanced quad through the intermediate-strip pipeline: it samples
/// `source()` of the page in [`parity`](Self::parity) and writes
/// [`bounds`](Self::bounds), modulated by [`opacity`](Self::opacity). The page
/// is free for reuse once the round holding this composite ends.
#[derive(Debug, Clone, PartialEq)]
pub struct Composite {
    /// The layer being composited, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    pub layer: u32,
    /// The group holding the finished page.
    pub parity: PageParity,
    /// Where the layer lands in the round's target, in that target's own
    /// coordinates.
    pub bounds: RectU16,
    /// Constant alpha applied while compositing, strictly between 0 and 1 for
    /// a pooled page; a retained page's is the bracket's own alpha, which may
    /// be 1.
    pub opacity: f32,
    /// The retained page this composite samples instead of a pooled one, or
    /// `None` for a pooled page. When set, [`parity`](Self::parity) names no
    /// page and nothing is released for it.
    pub retained: Option<RetainedSource>,
}

impl Composite {
    /// The region of the page to sample, in page texels.
    ///
    /// The layer was rendered at the page's origin, so this is
    /// [`bounds`](Self::bounds) moved there — or, for a retained page drawn
    /// trimmed to its window, moved to the texel its trimmed corner samples.
    #[must_use]
    pub fn source(&self) -> RectU16 {
        let (x, y) = self.retained.map_or((0, 0), |source| source.origin);
        RectU16::new(
            x,
            y,
            x.saturating_add(self.bounds.width()),
            y.saturating_add(self.bounds.height()),
        )
    }
}

/// Where a retained composite's page comes from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RetainedSource {
    /// The retained bracket the page belongs to, indexed into the frame's
    /// [`CompiledFrame::retained`](crate::compile::CompiledFrame::retained) —
    /// the renderer resolves it to the store's texture.
    pub bracket: u32,
    /// The page texel under the composite's top-left corner.
    pub origin: (u16, u16),
}

/// A retained page a round renders a recorded body into (see the module
/// header's *Retained pages*).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RetainedTarget {
    /// The body's layer, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    pub layer: u32,
    /// The retained bracket the page belongs to (see
    /// [`RetainedSource::bracket`]).
    pub bracket: u32,
    /// The body's tile-aligned device bounds, held at the page's origin.
    pub bounds: RectU16,
    /// Whether an earlier round of this body already rendered into the page
    /// this frame, so this one loads it instead of clearing it.
    pub continued: bool,
}

/// How the retained-page store resolved one recorded layer standing for a
/// retained bracket.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RetainedRole {
    /// A body rendered into its retained page and composited from it.
    Record {
        /// The retained bracket (see [`RetainedSource::bracket`]).
        bracket: u32,
    },
    /// A `DrawRetained` composited from a resident page.
    Draw {
        /// The retained bracket (see [`RetainedSource::bracket`]).
        bracket: u32,
        /// Where the composite lands.
        dest: RectU16,
        /// The page texel under `dest`'s top-left corner.
        origin: (u16, u16),
    },
    /// A `DrawRetained` whose composite lands nowhere visible: nothing at all.
    Hidden,
}

/// The [`RetainedRole`] of every recorded layer the retained-page store
/// resolved, indexed by layer id; a layer it did not is scheduled like any
/// other.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RetainedLayers {
    roles: Vec<Option<RetainedRole>>,
}

impl RetainedLayers {
    /// No layer resolved: [`Schedule::build_retained`] schedules exactly as
    /// [`Schedule::build`] does.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Forget every role, keeping the allocation.
    pub fn clear(&mut self) {
        self.roles.clear();
    }

    /// Give layer `layer` its role.
    pub fn set(&mut self, layer: u32, role: RetainedRole) {
        let index = layer as usize;
        if self.roles.len() <= index {
            self.roles.resize(index.saturating_add(1), None);
        }
        if let Some(slot) = self.roles.get_mut(index) {
            *slot = Some(role);
        }
    }

    /// Layer `layer`'s role, or `None` for an ordinary layer.
    #[must_use]
    pub fn role(&self, layer: u32) -> Option<RetainedRole> {
        self.roles.get(layer as usize).copied().flatten()
    }
}

/// One pass of a filter layer's sequence, reading one page and writing the
/// other.
///
/// One instanced quad through the filter pipeline. The page it writes is the
/// round's own target, so only the page it reads is named here; the extents it
/// reads and writes are [`step`](Self::step)'s, both taken at their page's
/// origin because a layer is always rendered there.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FilterPass {
    /// The layer being filtered, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    pub layer: u32,
    /// Which pass of the filter's sequence this is, and at what extents.
    pub step: FilterStep,
    /// The group holding the page this pass reads.
    pub source: PageParity,
}

/// One unit of work inside a round, executed in list order.
#[derive(Debug, Clone, PartialEq)]
pub enum RoundOp {
    /// Draws to issue into the round's target, indexed into
    /// [`CommandRecorder::draws`](vello_common::record::CommandRecorder::draws).
    ///
    /// Never empty: a node that recorded no draw contributes no op.
    Draws(Range<u32>),
    /// A finished page composited onto the round's target.
    Composite(Composite),
}

/// One render pass over one target.
///
/// A target is not a round: the surface, and a layer's page, can each take
/// several, and the rounds of one target run in the order they are listed with
/// every one after the first loading what the one before it left. The
/// surface's first round that draws is the one that clears it — the renderer
/// folds the frame's clear and its opaque strips into that round's pass.
#[derive(Debug, Clone, PartialEq)]
pub struct Round {
    /// What this pass renders into.
    pub target: RoundTarget,
    /// The pass's work, in execution order. Empty on a filter round, which
    /// issues no draw and composites nothing.
    pub ops: Vec<RoundOp>,
    /// Pages whose contents this round consumed; they return to the pool once
    /// it completes.
    pub released: Vec<PageParity>,
    /// The filter pass this round runs, on a filter round.
    ///
    /// A field rather than a [`RoundOp`] variant, and it stays one now that the
    /// renderer executes filter rounds: a filter round is *only* a filter pass
    /// — it draws nothing and composites nothing, and it is drawn through a
    /// different pipeline, off a different instance buffer, with no viewport
    /// uniform and no strip bind groups. Naming it here is what lets the frame
    /// path branch on the round before it starts issuing a list, and keeps
    /// [`RoundOp`] exactly the set of things one strip pass can interleave.
    pub filter: Option<FilterPass>,
    /// The region of the frame's own target this round seeds its page with,
    /// on a region-copy round (see the module header's *Backdrop rounds*).
    ///
    /// A field rather than a [`RoundOp`] for the reason [`Self::filter`] is
    /// one: a copy round is *only* a copy — it draws nothing, composites
    /// nothing, and is recorded through no strip pipeline — and its `ops` are
    /// empty.
    pub copy: Option<RegionCopy>,
    /// The pages this round's draws *sample* rather than composite: one per
    /// backdrop blur whose rectangle this round draws, each holding the
    /// blurred region (see the module header's *Backdrop rounds*). Every page
    /// named here is also in [`Self::released`]. Empty on every round of a
    /// frame with no backdrop, and on every round but a surface one.
    pub samples: Vec<BackdropSample>,
}

/// A region of the frame's own target, copied into a round's page at the
/// page's origin before anything else reads the page.
///
/// The reusable half of a backdrop blur: whatever needs to read what the frame
/// has already painted — a blur of it, and next a blend composite against it —
/// is handed a page holding exactly that region, made by cutting the surface
/// round so everything before it in painter order has landed, then copying.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RegionCopy {
    /// The device-space region copied, which is also the page's own
    /// [`PageTarget::bounds`]: the page holds it at its origin.
    pub source: RectU16,
}

/// A finished page a surface round's draws sample at their own position — the
/// blurred backdrop a backdrop blur's rectangle is painted from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BackdropSample {
    /// The backdrop's recorded layer, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    pub layer: u32,
    /// The group holding the page.
    pub parity: PageParity,
}

impl Round {
    /// The pooled page this round renders into, or `None` for the surface
    /// and for a retained page.
    #[must_use]
    pub fn page(&self) -> Option<&PageTarget> {
        match &self.target {
            RoundTarget::Root | RoundTarget::Retained(_) => None,
            RoundTarget::Page(page) => Some(page),
        }
    }

    /// The retained page this round renders a body into, if it does.
    #[must_use]
    pub fn retained(&self) -> Option<&RetainedTarget> {
        match &self.target {
            RoundTarget::Retained(target) => Some(target),
            RoundTarget::Root | RoundTarget::Page(_) => None,
        }
    }

    /// The filter pass this round runs, or `None` on an ordinary round.
    #[must_use]
    pub fn filter_pass(&self) -> Option<&FilterPass> {
        self.filter.as_ref()
    }

    /// The region this round copies into its page, or `None` on every round
    /// but a region-copy one.
    #[must_use]
    pub fn region_copy(&self) -> Option<&RegionCopy> {
        self.copy.as_ref()
    }

    /// Whether this round renders into the frame's own surface.
    #[must_use]
    pub fn is_root(&self) -> bool {
        matches!(self.target, RoundTarget::Root)
    }

    /// The composites this round performs, in execution order.
    pub fn composites(&self) -> impl Iterator<Item = &Composite> {
        self.ops.iter().filter_map(|op| match op {
            RoundOp::Composite(composite) => Some(composite),
            RoundOp::Draws(_) => None,
        })
    }

    /// How many recorded draws this round issues.
    #[must_use]
    pub fn draw_count(&self) -> u32 {
        self.ops
            .iter()
            .map(|op| match op {
                RoundOp::Draws(range) => range.end.saturating_sub(range.start),
                RoundOp::Composite(_) => 0,
            })
            .sum()
    }
}

/// Turns a recording into the rounds that render it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Schedule;

impl Schedule {
    /// The rounds `recorder` renders as: every layer's rounds before the round
    /// that composites it, and the surface's last round last.
    ///
    /// Always at least one round: a recording with no layers, and one whose
    /// layers all inline or drop away, both schedule to the single root round
    /// that clears and paints the surface.
    ///
    /// # Errors
    ///
    /// [`EngineError::SchedulerEscalation`] for a layer shape outside what this
    /// scheduler serves (see the module header), with a reason naming what was
    /// found, and [`EngineError::IntermediateTextureTooLarge`] for a layer
    /// larger than a page can be sized to. Neither is a panic, and neither
    /// routes the frame anywhere: both hand the caller a frame it is expected
    /// to skip.
    ///
    /// A Gaussian-blur or shadow-only drop-shadow filter layer is *planned*
    /// here: its contents' round, then one round per pass of its filter's
    /// sequence, then the composite of whichever page the last pass wrote.
    /// Every other filter is refused by name (see
    /// [`crate::filters::served_filter`]), and so are the two shapes a served
    /// filter layer still cannot take — one recorded inside another layer, and
    /// one that cannot be handed the second page group its passes ping-pong
    /// into.
    ///
    /// One entry point, not two. While the renderer had no filter pipeline,
    /// this call refused every filter and a second one planned them, so the
    /// frame path could not reach a round nothing could execute; the renderer
    /// runs them now ([`crate::renderer::FilterResources`]), and the split went
    /// with the reason for it.
    pub fn build(
        recorder: &CommandRecorder<EngineDraw>,
        caps: &TierCaps,
        config: &PageConfig,
    ) -> Result<Vec<Round>, EngineError> {
        Self::build_retained(recorder, caps, config, &RetainedLayers::new())
    }

    /// [`Self::build`], with the layers the retained-page store resolved
    /// served as `retained` says (see the module header's *Retained pages*).
    ///
    /// # Errors
    ///
    /// Exactly [`Self::build`]'s. A retained body is held to the same layer
    /// properties every layer is (no blend mode, mask or clip path, a finite
    /// opacity) and its nested layers to every bound; the body itself takes no
    /// page group, so it can never be the layer a shortage refuses.
    pub fn build_retained(
        recorder: &CommandRecorder<EngineDraw>,
        caps: &TierCaps,
        config: &PageConfig,
        retained: &RetainedLayers,
    ) -> Result<Vec<Round>, EngineError> {
        Self::build_frame(recorder, caps, config, retained, &[])
    }

    /// [`Self::build_retained`], with the frame's backdrop blurs served as the
    /// module header's *Backdrop rounds* describes. With no backdrop this is
    /// exactly [`Self::build_retained`]: nothing is cut, copied or sampled.
    ///
    /// # Errors
    ///
    /// Exactly [`Self::build`]'s. A backdrop's page is sized by
    /// [`filter_page_size`] like a filter layer's, so a region larger than a
    /// page can be is refused with
    /// [`EngineError::IntermediateTextureTooLarge`] on the same terms.
    pub fn build_frame(
        recorder: &CommandRecorder<EngineDraw>,
        caps: &TierCaps,
        config: &PageConfig,
        retained: &RetainedLayers,
        backdrops: &[Backdrop],
    ) -> Result<Vec<Round>, EngineError> {
        let mut rounds: Vec<Round> = Vec::new();
        let mut pages = LivePages::default();
        // A child layer is always recorded after its parent, so a well-formed
        // recording enters each layer exactly once. Marking them is what keeps
        // a recording whose nodes re-enter one from being walked forever, and
        // it bounds the stack below to one entry per recorded layer.
        let mut entered = vec![false; recorder.layers.len()];
        let mut stack = vec![Stream::root(&recorder.nodes)];

        while !stack.is_empty() {
            // Taking one node off the innermost open stream is a borrow of the
            // stack that has to end before the step below can push or pop it.
            let step = match stack.last_mut() {
                None => break,
                Some(stream) => match stream.nodes.get(stream.index) {
                    None => Step::Done,
                    Some(node) => {
                        stream.index = stream.index.saturating_add(1);
                        if node.draws.start < node.draws.end {
                            stream.ops.push(RoundOp::Draws(node.draws.clone()));
                        }
                        match node.layer {
                            None => Step::Next,
                            Some(id) => Step::Enter {
                                id,
                                depth: stream.depth,
                            },
                        }
                    }
                },
            };

            match step {
                Step::Next => {}
                Step::Done => {
                    if let Some(stream) = stack.pop() {
                        let carry =
                            finish(stream, &mut stack, &mut rounds, &mut pages, caps, config)?;
                        if let Some(parent) = stack.last_mut() {
                            parent.ops.extend(carry);
                        }
                    }
                }
                Step::Enter { id, depth } => {
                    let index = id as usize;
                    let (Some(layer), Some(seen)) =
                        (recorder.layers.get(index), entered.get_mut(index))
                    else {
                        return Err(escalate(format!(
                            "a node enters layer {id}, which the recording does not hold"
                        )));
                    };
                    if *seen {
                        return Err(escalate(format!(
                            "layer {id} is entered a second time; the simple scheduler walks \
                             each recorded layer once, and a recording whose nodes re-enter one \
                             describes no tree it could render"
                        )));
                    }
                    *seen = true;

                    // Suppression is inherited rather than re-derived: a layer
                    // inside one covering no pixels covers none either, however
                    // its own bounds read.
                    let inherited = stack.last().is_some_and(|stream| stream.suppressed);

                    if let Some(backdrop) = backdrops.iter().find(|backdrop| backdrop.layer == id) {
                        // The layer properties every layer is held to; a
                        // backdrop's own are always the inlining ones.
                        layer_role(id, layer)?;
                        // Served only where the frame's own target is what the
                        // walk is drawing into — the root, or layers inlined
                        // into it. Inside an isolated layer (or a retained
                        // body, or one covering nothing) there is no frame
                        // beneath to sample: the backdrop and its rectangle
                        // draw nothing, and never land in a page round.
                        let at_root = stack
                            .iter()
                            .rev()
                            .find(|stream| !matches!(stream.owner, StreamOwner::Inline))
                            .is_some_and(|stream| matches!(stream.owner, StreamOwner::Root));
                        if at_root && !inherited && !layer.bbox.is_empty() {
                            backdrop_rounds(
                                backdrop,
                                &mut stack,
                                &mut rounds,
                                &mut pages,
                                caps,
                                config,
                            )?;
                            stack.push(Stream::inline(layer, depth, false));
                        }
                        continue;
                    }

                    if let Some(role) = retained.role(id) {
                        // The layer properties every layer is held to; the
                        // isolation answer is the store's, not opacity's.
                        layer_role(id, layer)?;
                        match role {
                            RetainedRole::Record { bracket } => {
                                stack.push(Stream::retained(id, layer, bracket, depth, inherited))
                            }
                            RetainedRole::Draw {
                                bracket,
                                dest,
                                origin,
                            } => {
                                if !inherited
                                    && layer.props.opacity > 0.0
                                    && let Some(stream) = stack.last_mut()
                                {
                                    stream.ops.push(RoundOp::Composite(Composite {
                                        layer: id,
                                        parity: PageParity::Even,
                                        bounds: dest,
                                        opacity: layer.props.opacity,
                                        retained: Some(RetainedSource { bracket, origin }),
                                    }));
                                }
                            }
                            RetainedRole::Hidden => {}
                        }
                        continue;
                    }

                    // Nothing nested inside a dropped layer is rendered, so
                    // nothing inside one is validated either — the walk does
                    // not enter it.
                    let role = layer_role(id, layer)?;
                    // A filter layer's placement in its parent is computed by
                    // undoing a shift the reference renderer applies to a
                    // filter layer's contents and frust's compiler does not, so
                    // the bounds it hands its parent fall short of the filter's
                    // own spread on two sides. Refused rather than rendered
                    // from bounds that would clip it; a backdrop blur is
                    // recorded under the surface, where nothing reads that
                    // placement.
                    if matches!(role, LayerRole::Filtered) && stack.len() > 1 {
                        return Err(escalate(format!(
                            "filter layer {id} is recorded inside another layer, whose own bounds \
                             would be taken from a placement that undoes a source shift this \
                             engine's compiler never applied; a filter layer is served directly \
                             under the frame's own surface"
                        )));
                    }

                    match role {
                        LayerRole::Dropped => {}
                        LayerRole::Inline => stack.push(Stream::inline(layer, depth, inherited)),
                        LayerRole::Isolated | LayerRole::Filtered => {
                            let depth = depth.saturating_add(1);
                            if depth > MAX_CHAIN_DEPTH {
                                return Err(escalate(format!(
                                    "{depth} nested isolated layers, deeper than the \
                                     {MAX_CHAIN_DEPTH}-deep chain the simple scheduler serves"
                                )));
                            }
                            // A layer whose contents cover nothing composites
                            // nothing, so it renders no round — and neither do
                            // its descendants, whose pages nothing would ever
                            // sample. The walk still enters it, so a shape this
                            // scheduler refuses is refused wherever it is
                            // recorded rather than only where it is visible.
                            let suppressed = inherited || layer.bbox.is_empty();
                            stack.push(Stream::isolated(id, layer, depth, suppressed));
                        }
                    }
                }
            }
        }

        Ok(rounds)
    }
}

/// Emits the last round a finished stream renders as, and returns what its
/// parent splices in at the node that entered it: one composite for an isolated
/// layer, the layer's own ops for an inlined one, nothing for a layer that
/// composites nothing. The frame root pushes the last round and carries nothing.
///
/// `stack` is the walk's remaining streams — this stream's ancestors — because
/// a layer short of a page makes room by [cutting](cut_at) one of their rounds.
fn finish<'a>(
    stream: Stream<'a>,
    stack: &mut [Stream<'a>],
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<Vec<RoundOp>, EngineError> {
    let (id, layer) = match stream.owner {
        StreamOwner::Inline => return Ok(stream.ops),
        StreamOwner::Root => {
            let mut released = released_pages(&stream.ops);
            release_samples(&stream.samples, &mut released);
            for parity in &released {
                pages.release(*parity);
            }
            // An empty trailing round is worth a pass only when it is the
            // frame's only one: `build` always returns at least one round, but
            // a surface already written by an earlier cut needs no empty pass
            // behind it.
            if !stream.ops.is_empty() || !stream.emitted {
                rounds.push(Round {
                    target: RoundTarget::Root,
                    ops: stream.ops,
                    released,
                    filter: None,
                    copy: None,
                    samples: stream.samples,
                });
            }
            return Ok(Vec::new());
        }
        StreamOwner::Retained { id, layer, bracket } => {
            return Ok(finish_retained(stream, id, layer, bracket, rounds, pages));
        }
        StreamOwner::Isolated { id, layer } => (id, layer),
    };

    // A layer covering no pixels, or one nested inside such a layer, renders
    // nothing: no round of its own was cut, none of its descendants' was
    // emitted, and no page was ever taken for any of them, so there is nothing
    // to release and nothing for the parent to composite.
    if stream.suppressed {
        return Ok(Vec::new());
    }

    let bounds = layer.bbox;
    let depth = stream.depth;

    // A regular layer wider than any single page is banded into column
    // pages (E14) rather than refused — but only while nothing has claimed
    // its page ahead of time (a cut ancestor reserves one this way, see
    // `cut_at`) and its own accumulated ops hold no composite of a nested
    // isolated child: a band's ops are replayed once per band, and
    // replaying a child's composite would read a page a later band has
    // already reused. Both are shapes banding does not reach, and a layer
    // whose height alone exceeds the ceiling is still refused by `page_size`
    // below exactly as it always was — bands are columns, so height has no
    // split to be served by.
    if matches!(layer.kind, RecordedLayerKind::Regular)
        && stream.page.is_none()
        && u32::from(bounds.width()) > page_ceiling(config, caps)
        && !holds_composite(&stream.ops)
    {
        return band_rounds(
            BandedLayer {
                id,
                layer,
                bounds,
                depth,
            },
            stream.ops,
            stack,
            rounds,
            pages,
            caps,
            config,
        );
    }

    // A filter layer's page is sized with the extra atlas margin
    // (`filter_page_size`) its kernel taps can reach past: both this
    // contents round and every filter-pass round the layer costs share this
    // one `size` (see `filter_rounds`'s own `filtered.size`), so the margin
    // has to be reserved here, before the first byte of the layer is ever
    // rendered — sizing only the pass rounds would leave the contents
    // themselves on an unpadded page while the first pass reads past its
    // edge. A regular layer never filters its contents, so it keeps the
    // plain `page_size` this always used.
    let size = match &layer.kind {
        RecordedLayerKind::Regular => page_size(bounds, config, caps)?,
        _ => filter_page_size(bounds, config, caps)?,
    };
    let page = |parity, continued| {
        RoundTarget::Page(PageTarget {
            layer: id,
            depth,
            parity,
            size,
            bounds,
            continued,
        })
    };

    // The group holding the layer's own contents once its last content round
    // has been emitted.
    let contents = match stream.page {
        // The layer's round was cut earlier this frame, so its page is already
        // this stream's and this round loads what the cut left in it.
        Some(parity) => {
            if !stream.ops.is_empty() {
                let released = released_pages(&stream.ops);
                for parity in &released {
                    pages.release(*parity);
                }
                rounds.push(Round {
                    target: page(parity, true),
                    ops: stream.ops,
                    released,
                    filter: None,
                    copy: None,
                    samples: Vec::new(),
                });
            }
            parity
        }
        None => {
            make_room(stack, rounds, pages, caps, config)?;
            // Acquired before the children's pages are freed, never after: a
            // child's page is live for the whole of the pass that samples it,
            // so the group this round renders into can never be one of theirs.
            // With both groups still held after `make_room` has cut everything
            // it could, a regular layer takes the one spill page rather than
            // refusing the frame — see the module header's *The spill page*.
            let spills = matches!(layer.kind, RecordedLayerKind::Regular);
            let parity = pages
                .acquire(PageParity::from_depth(depth))
                .or_else(|| spills.then(|| pages.acquire_spill()).flatten())
                .ok_or_else(|| escalate(out_of_pages(id, spills)))?;
            let released = released_pages(&stream.ops);
            for parity in &released {
                pages.release(*parity);
            }
            rounds.push(Round {
                target: page(parity, false),
                ops: stream.ops,
                released,
                filter: None,
                copy: None,
                samples: Vec::new(),
            });
            parity
        }
    };

    // A filter layer's contents are not what the parent composites: the filter
    // passes run over them first, and the parent samples whatever the last of
    // them wrote.
    let composited = match &layer.kind {
        RecordedLayerKind::Regular => contents,
        kind => filter_rounds(
            &FilterLayer {
                id,
                kind,
                depth,
                bounds,
                size,
                contents,
            },
            stack,
            rounds,
            pages,
            caps,
            config,
        )?,
    };

    Ok(vec![RoundOp::Composite(Composite {
        layer: id,
        parity: composited,
        bounds,
        opacity: layer.props.opacity,
        retained: None,
    })])
}

/// Emits the last round a retained body renders as — into the store's page,
/// never a pooled one — and returns the composite of that page for the parent
/// to splice in (nothing at zero opacity: the page is still wanted, but this
/// frame shows none of it).
///
/// A body that renders anything renders at least one round even when it is
/// empty of ops: that round is what clears the page, and a page the store
/// believes rendered must hold this frame's body rather than an earlier one.
/// A body nested inside a layer covering nothing renders no round at all, and
/// the store, seeing none, leaves its key as it was.
fn finish_retained(
    stream: Stream<'_>,
    id: u32,
    layer: &RecordedLayer,
    bracket: u32,
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
) -> Vec<RoundOp> {
    if stream.suppressed {
        return Vec::new();
    }
    let released = released_pages(&stream.ops);
    for parity in &released {
        pages.release(*parity);
    }
    if !stream.ops.is_empty() || !stream.emitted {
        rounds.push(Round {
            target: RoundTarget::Retained(RetainedTarget {
                layer: id,
                bracket,
                bounds: layer.bbox,
                continued: stream.emitted,
            }),
            ops: stream.ops,
            released,
            filter: None,
            copy: None,
            samples: Vec::new(),
        });
    }

    let opacity = layer.props.opacity;
    if opacity > 0.0 {
        vec![RoundOp::Composite(Composite {
            layer: id,
            parity: PageParity::Even,
            bounds: layer.bbox,
            opacity,
            retained: Some(RetainedSource {
                bracket,
                origin: (0, 0),
            }),
        })]
    } else {
        Vec::new()
    }
}

/// A filter layer whose contents are rendered, ready for its passes to be
/// planned.
struct FilterLayer<'a> {
    /// The layer, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    id: u32,
    /// The recorded filter, re-read here rather than carried down from
    /// [`layer_role`]: preparing a blur's kernel is a handful of arithmetic
    /// over a fixed-size array, and threading it through the walk's stack would
    /// put a filter's state in every stream that has none.
    kind: &'a RecordedLayerKind,
    /// The layer's depth counted over isolated layers only, one-based.
    depth: usize,
    /// The layer's tile-aligned device-space bounds, already grown by the
    /// filter's own spread (see [`crate::filters::LayerFilter::filter_data`]).
    bounds: RectU16,
    /// The extent both of the layer's pages are acquired at.
    size: PageSize,
    /// The group the layer's own contents were rendered into.
    contents: PageParity,
}

/// Emits the rounds a filter layer's pass sequence renders as, and answers the
/// group holding the filtered result for its parent to composite.
///
/// One round per pass, each writing the group it did not read. The sequence is
/// even in length, so the result lands back in the page the contents were
/// rendered into; [`blur::blur_passes`] is what keeps that true.
fn filter_rounds(
    filtered: &FilterLayer<'_>,
    stack: &mut [Stream<'_>],
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<PageParity, EngineError> {
    let id = filtered.id;
    // The one dispatch every filter-recognising site in this crate shares
    // (`renderer::filter_block`, `layer_role` below): the recorded primitive
    // names which filter it is, and the reason a refusal carries is that
    // primitive's own.
    let steps = match served_filter(id, filtered.kind).map_err(escalate)? {
        ServedFilter::Blur(blur) => blur::blur_passes(&blur, SizeU16::from(filtered.bounds)),
        ServedFilter::DropShadow(shadow) => {
            drop_shadow::drop_shadow_passes(&shadow, SizeU16::from(filtered.bounds))
        }
    };
    pass_rounds(filtered, &steps, stack, rounds, pages, caps, config)
}

/// Emits one round per pass of `steps` over `filtered`'s pages, and answers
/// the group holding the result — [`filter_rounds`]' own sequence, shared with
/// a backdrop blur's ([`backdrop_rounds`]), whose page is seeded by a region
/// copy instead of by rendered contents.
fn pass_rounds(
    filtered: &FilterLayer<'_>,
    steps: &[FilterStep],
    stack: &mut [Stream<'_>],
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<PageParity, EngineError> {
    let id = filtered.id;
    let Some(last) = steps.len().checked_sub(1) else {
        return Ok(filtered.contents);
    };

    // A filter layer holds both groups from its first pass to its last, so an
    // open round still holding the other one is cut to hand it back — innermost
    // first, and only until it comes back. A cut cannot take the group instead:
    // with both held there is none for it to acquire.
    let scratch = filtered.contents.opposite();
    for index in (0..stack.len()).rev() {
        if !pages.holds(scratch) {
            break;
        }
        cut_at(stack, index, rounds, pages, caps, config)?;
    }
    let scratch = pages.acquire(scratch).ok_or_else(|| {
        escalate(format!(
            "filter layer {id} would need both of the {PING_PONG_GROUPS} page groups at once — \
             one for its contents and one for its passes to write — and the second is holding a \
             page a later round composites that no open round could be cut to hand back"
        ))
    })?;

    let mut source = filtered.contents;
    let mut dest = scratch;
    for (index, step) in steps.iter().enumerate() {
        // The page a pass reads is the page the pass before it wrote, so both
        // stay live for the whole sequence. The scratch one goes back to the
        // pool as the last pass ends; the one holding the result goes back when
        // the parent's composite has sampled it.
        let released = if index == last {
            vec![source]
        } else {
            Vec::new()
        };
        for parity in &released {
            pages.release(*parity);
        }

        rounds.push(Round {
            // Cleared, never continued: a pass writes only the region its step
            // names — a decimated one a quarter of the texels the pass before
            // it did — and the kernels sample past that region without bounds
            // checks, so what surrounds it has to be transparent rather than a
            // previous holder's pixels.
            target: RoundTarget::Page(PageTarget {
                layer: id,
                depth: filtered.depth,
                parity: dest,
                size: filtered.size,
                bounds: filtered.bounds,
                continued: false,
            }),
            ops: Vec::new(),
            released,
            filter: Some(FilterPass {
                layer: id,
                step: *step,
                source,
            }),
            copy: None,
            samples: Vec::new(),
        });

        if index < last {
            core::mem::swap(&mut source, &mut dest);
        }
    }

    Ok(dest)
}

/// A regular layer wider than [`page_ceiling`] can size a single page to,
/// ready for its column bands to be planned.
struct BandedLayer<'a> {
    /// The layer, indexed into
    /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
    id: u32,
    /// The recorded layer, for its opacity.
    layer: &'a RecordedLayer,
    /// The layer's tile-aligned device-space bounds — wider than
    /// [`page_ceiling`], which is why it is here rather than sized by
    /// [`page_size`] like every other regular layer.
    bounds: RectU16,
    /// The layer's depth counted over isolated layers only, one-based.
    depth: usize,
}

/// Emits the rounds a banded layer's column pages render as: one page and
/// one composite per [band](pages::page_bands), the layer's own draws
/// replayed into each.
///
/// A band is rendered and composited one at a time rather than the whole split
/// acquired up front, because a live page is scarce here: each band's composite
/// is spliced straight onto `stack`'s own innermost accumulator (exactly where
/// an ordinary single-page layer's composite would land), `make_room` is asked
/// again before the next band's page is acquired, and the same cutting it
/// already does for a wide sibling fan is what hands a finished band's page
/// back — a banded layer is, from the pool's point of view, a fan of same-depth
/// siblings that happen to share one layer id. See the [`pages`] module
/// header's *A fourth decision* section for why this is a page decision rather
/// than a new kind of target.
///
/// `ops` is `banded`'s stream's own accumulated ops — verified by the caller
/// to hold no [`RoundOp::Composite`] — replayed unchanged into every band's
/// content round; only the page each round targets and the rectangle its
/// composite lands at differ band to band.
///
/// Replayed unchanged, and *clipped* on the way to the GPU: every band is
/// offered the whole layer's draws, and each band's own
/// [`PageTarget::bounds`] is what selects the strips that belong to it — a
/// strip left of the band's column contributes no instance to it, and one
/// straddling the column's edge is advanced into it rather than clamped onto
/// it. That is what makes compositing the bands in order paint exactly what one
/// whole-layer page would have; the clip itself is the renderer's, at instance
/// emission, since the ops here are draw ranges rather than geometry.
///
/// A band takes a ping-pong group or nothing: the [spill
/// page](PageParity::Spill) is deliberately not offered here. It exists for a
/// layer whose own round samples a live page while an ancestor holds another,
/// and a banded layer is reached only when its ops hold no composite at all —
/// so the previous band's composite is always sitting in an open round
/// `make_room` can cut, and a third page is never what a band is short of.
///
/// # Errors
///
/// Whatever [`pages::page_bands`] itself refuses `banded.bounds` with: taller
/// than [`page_ceiling`] (bands are columns, so height has no split to be
/// served by), or wider than [`pages::MAX_PAGE_BANDS`] bands can cover. Both
/// are the refusal a single page gives an over-ceiling layer, unchanged by
/// banding. [`EngineError::SchedulerEscalation`] when a band's own page
/// cannot be found even one at a time.
fn band_rounds(
    banded: BandedLayer<'_>,
    ops: Vec<RoundOp>,
    stack: &mut [Stream<'_>],
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<Vec<RoundOp>, EngineError> {
    let bands = pages::page_bands(banded.bounds, config, caps)?;

    for band in bands {
        make_room(stack, rounds, pages, caps, config)?;
        // Acquired before anything of this band is released, never after —
        // the same ordering a single-page layer's own acquisition keeps.
        let parity = pages
            .acquire(PageParity::from_depth(banded.depth))
            .ok_or_else(|| {
                escalate(format!(
                    "layer {} would need a third live intermediate page to render one of its own \
                     column bands: both of the {PING_PONG_GROUPS} groups the simple scheduler \
                     ping-pongs between are already holding a page a later round composites, and \
                     no open round could be cut to hand one back",
                    banded.id
                ))
            })?;

        rounds.push(Round {
            target: RoundTarget::Page(PageTarget {
                layer: banded.id,
                depth: banded.depth,
                parity,
                size: band.size,
                bounds: band.bounds,
                // Every band is a page of its own rather than a continuation
                // of the one before it: two bands never share a texture at
                // once, so there is nothing here for a later band to load.
                continued: false,
            }),
            ops: ops.clone(),
            released: Vec::new(),
            filter: None,
            copy: None,
            samples: Vec::new(),
        });

        // Spliced onto the enclosing stream's own accumulator directly,
        // rather than returned for the caller to append: a later band still
        // has to acquire a group, and `make_room`'s own cutting is what finds
        // this composite and hands the group back — exactly the mechanism a
        // page-hungry sibling already relies on, so nothing here needs to
        // wait for this function to return before a round can use it.
        if let Some(parent) = stack.last_mut() {
            parent.ops.push(RoundOp::Composite(Composite {
                layer: banded.id,
                parity,
                bounds: band.bounds,
                opacity: banded.layer.props.opacity,
                retained: None,
            }));
        }
    }

    Ok(Vec::new())
}

/// Frees a page group for a layer about to take one, by cutting an ancestor's
/// round short, when the walk is running out of groups.
///
/// Two moves, in this order.
///
/// - **Ahead of the shortage.** An isolated ancestor that has not taken a page
///   yet is going to need one, and a layer taking the last free group would
///   leave it none — which is what used to limit a parent to two isolated
///   children. Cutting such an ancestor now gives it its page *and* hands back
///   every page that cut composites, so the group this layer is about to take
///   comes straight back and the count is no worse than it started. Outermost
///   first: an outer ancestor's cut is what frees the group an inner one takes.
/// - **At the shortage.** With both groups gone, the innermost open round that
///   composites anything is cut, releasing its pages a pass early. This is what
///   serves a fan of any width under the surface: the siblings already
///   composited go back to the pool before the next one is rendered.
///
/// A cut that cannot help is skipped rather than forced, and running out of
/// ancestors to cut is not itself an error — the acquire this precedes
/// escalates if it still finds no group, which keeps one refusal site and one
/// reason.
fn make_room(
    stack: &mut [Stream<'_>],
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<(), EngineError> {
    if pages.free() > 1 {
        return Ok(());
    }

    for index in 0..stack.len() {
        if pages.free() == 0 {
            break;
        }
        if stack[index].wants_a_page() {
            cut_at(stack, index, rounds, pages, caps, config)?;
        }
    }

    if pages.free() == 0 {
        for index in (0..stack.len()).rev() {
            if cut_at(stack, index, rounds, pages, caps, config)? {
                break;
            }
        }
    }

    Ok(())
}

/// Emits what `stack[index]` has accumulated as a round of its own, so the pages
/// that batch composites return to the pool before the walk renders anything
/// else. Answers whether a round was emitted.
///
/// The round carries that stream's ops followed by those of every inlined stream
/// open above it: an inlined layer draws into its host's round, so its ops
/// belong to this cut, in exactly the painter order the recording laid them in.
/// Whatever any of them records after the cut becomes the host's next round,
/// which loads rather than clears (see [`PageTarget::continued`]) — so a cut
/// changes when a target's pixels are written, never which ones or in what
/// order.
///
/// `false` when there is nothing to gain or no way to emit: a batch holding no
/// composite frees no page by being cut, a suppressed stream renders nothing at
/// all, an inlined one has no target of its own, and an isolated one that has
/// not taken a page cannot take one with both groups held.
fn cut_at(
    stack: &mut [Stream<'_>],
    index: usize,
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<bool, EngineError> {
    let Some(stream) = stack.get(index) else {
        return Ok(false);
    };
    if stream.suppressed || matches!(stream.owner, StreamOwner::Inline) {
        return Ok(false);
    }

    // Every stream above this one up to the next stream with a target of its own
    // is inlined into this round.
    let end = stack
        .iter()
        .enumerate()
        .skip(index.saturating_add(1))
        .find(|(_, stream)| !matches!(stream.owner, StreamOwner::Inline))
        .map_or(stack.len(), |(at, _)| at);
    let batch = stack.get(index..end).unwrap_or(&[]);
    if !batch
        .iter()
        .any(|stream| holds_composite(&stream.ops) || !stream.samples.is_empty())
    {
        return Ok(false);
    }

    let continued = stack[index].emitted;
    let target = match stack[index].owner {
        // Both refused above; repeated here because the target is what the
        // round is built from.
        StreamOwner::Inline => return Ok(false),
        StreamOwner::Root => RoundTarget::Root,
        // A retained body's page is the store's, so cutting it takes no group:
        // the cut renders what the body has so far, and its next round loads
        // it.
        StreamOwner::Retained { id, layer, bracket } => RoundTarget::Retained(RetainedTarget {
            layer: id,
            bracket,
            bounds: layer.bbox,
            continued,
        }),
        StreamOwner::Isolated { id, layer } => {
            let bounds = layer.bbox;
            // A filter layer's stream can itself be cut mid-frame — nothing
            // stops a regular/opacity layer from being recorded nested inside
            // one, and `make_room` treats every isolated stream on the stack
            // alike — so this has to size the page on the same terms `finish`
            // does above, or a cut filter layer's contents would land on an
            // unpadded page its own first pass then reads past the edge of.
            let size = match &layer.kind {
                RecordedLayerKind::Regular => page_size(bounds, config, caps)?,
                _ => filter_page_size(bounds, config, caps)?,
            };
            let parity = match stack[index].page {
                Some(parity) => parity,
                None => {
                    let Some(parity) = pages.acquire(PageParity::from_depth(stack[index].depth))
                    else {
                        return Ok(false);
                    };
                    stack[index].page = Some(parity);
                    parity
                }
            };
            RoundTarget::Page(PageTarget {
                layer: id,
                depth: stack[index].depth,
                parity,
                size,
                bounds,
                continued,
            })
        }
    };

    let mut ops: Vec<RoundOp> = Vec::new();
    let mut samples: Vec<BackdropSample> = Vec::new();
    for stream in stack.get_mut(index..end).unwrap_or(&mut []) {
        ops.append(&mut stream.ops);
        samples.append(&mut stream.samples);
    }
    let mut released = released_pages(&ops);
    release_samples(&samples, &mut released);
    for parity in &released {
        pages.release(*parity);
    }
    stack[index].emitted = true;
    rounds.push(Round {
        target,
        ops,
        released,
        filter: None,
        copy: None,
        samples,
    });

    Ok(true)
}

/// Emits the rounds a backdrop blur at the frame root costs, leaving the page
/// holding the blurred region sampled by the root stream (see the module
/// header's *Backdrop rounds*).
///
/// Three steps, in order: the surface round open so far is emitted whatever it
/// holds, so everything painted before the backdrop has landed before anything
/// reads it back; a [region-copy](RegionCopy) round seeds a page with the
/// backdrop's region of the frame's target; and the blur's own pass sequence
/// runs over that page exactly as a filter layer's does. The result stays held
/// until the surface round that draws the backdrop's rectangle ends.
///
/// `stack` holds the frame root and nothing but inlined layers above it — the
/// caller serves a backdrop nowhere else — so cutting it leaves no page held
/// and both groups are free for the copy and the passes.
fn backdrop_rounds(
    backdrop: &Backdrop,
    stack: &mut [Stream<'_>],
    rounds: &mut Vec<Round>,
    pages: &mut LivePages,
    caps: &TierCaps,
    config: &PageConfig,
) -> Result<(), EngineError> {
    flush_surface(stack, rounds, pages);

    let id = backdrop.layer;
    let region = backdrop.region;
    let size = filter_page_size(region, config, caps)?;
    let contents = pages.acquire(PageParity::Even).ok_or_else(|| {
        escalate(format!(
            "backdrop layer {id} found no free page group for its region copy after the surface \
             round ahead of it was cut"
        ))
    })?;
    rounds.push(Round {
        // Cleared and then written by the copy: the page's previous holder
        // must not show past the copied region, where the kernels read.
        target: RoundTarget::Page(PageTarget {
            layer: id,
            depth: 1,
            parity: contents,
            size,
            bounds: region,
            continued: false,
        }),
        ops: Vec::new(),
        released: Vec::new(),
        filter: None,
        copy: Some(RegionCopy { source: region }),
        samples: Vec::new(),
    });

    let steps = blur::blur_passes(&backdrop.blur(), SizeU16::from(region));
    let regular = RecordedLayerKind::Regular;
    let blurred = pass_rounds(
        &FilterLayer {
            id,
            kind: &regular,
            depth: 1,
            bounds: region,
            size,
            contents,
        },
        &steps,
        stack,
        rounds,
        pages,
        caps,
        config,
    )?;

    if let Some(root) = stack.first_mut() {
        root.samples.push(BackdropSample {
            layer: id,
            parity: blurred,
        });
    }
    Ok(())
}

/// Emits everything the frame root and the inlined layers above it have
/// accumulated as a surface round of its own — unconditionally, unlike
/// [`cut_at`], which cuts only to hand a page back — or nothing when nothing
/// has accumulated since the last one.
fn flush_surface(stack: &mut [Stream<'_>], rounds: &mut Vec<Round>, pages: &mut LivePages) {
    let mut ops: Vec<RoundOp> = Vec::new();
    let mut samples: Vec<BackdropSample> = Vec::new();
    for stream in stack.iter_mut() {
        ops.append(&mut stream.ops);
        samples.append(&mut stream.samples);
    }
    if ops.is_empty() && samples.is_empty() {
        return;
    }
    let mut released = released_pages(&ops);
    release_samples(&samples, &mut released);
    for parity in &released {
        pages.release(*parity);
    }
    if let Some(root) = stack.first_mut() {
        root.emitted = true;
    }
    rounds.push(Round {
        target: RoundTarget::Root,
        ops,
        released,
        filter: None,
        copy: None,
        samples,
    });
}

/// Adds the pages `samples` holds to `released`, each listed once.
fn release_samples(samples: &[BackdropSample], released: &mut Vec<PageParity>) {
    for sample in samples {
        if !released.contains(&sample.parity) {
            released.push(sample.parity);
        }
    }
}

/// How a recorded layer is served.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LayerRole {
    /// Composites identically to drawing its contents directly, so its stream
    /// is spliced into its parent's round.
    Inline,
    /// Needs a page and a pass of its own.
    Isolated,
    /// Needs a page and a pass of its own, plus a second page and a pass per
    /// step of its filter's sequence.
    ///
    /// Never inlined however it composites: filtering a layer's contents is not
    /// what drawing them into the parent does.
    Filtered,
    /// Contributes nothing, along with everything nested inside it.
    Dropped,
}

/// What taking one node off the innermost open stream left the walk to do.
///
/// The step is decided under a borrow of the stack and acted on after it, since
/// entering a layer pushes a stream and finishing one pops it.
enum Step {
    /// The node entered a layer.
    Enter {
        /// The layer's index into
        /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
        id: u32,
        /// The isolated nesting depth of the stream that entered it.
        depth: usize,
    },
    /// The node entered nothing; the stream continues.
    Next,
    /// The stream is exhausted and renders as its round now.
    Done,
}

/// One command stream the walk has open: the frame root's own nodes, or one
/// recorded layer's.
///
/// The walk keeps these on an explicit stack rather than recursing, so a
/// recording nesting layers far past anything a widget tree records costs heap
/// rather than call frames. The stack is bounded by the recording's layer count,
/// because a layer is entered at most once.
struct Stream<'a> {
    /// Whose stream this is, and how it is served.
    owner: StreamOwner<'a>,
    /// The stream's nodes, in recording order.
    nodes: &'a [Node],
    /// The next node to visit.
    index: usize,
    /// The isolated nesting depth of this stream's contents, one-based; zero at
    /// the frame root.
    depth: usize,
    /// The ops accumulated since this stream's last round, in execution order.
    ops: Vec<RoundOp>,
    /// Whether a round of this stream's has been emitted already, which is what
    /// makes the next one a continuation of the same target.
    emitted: bool,
    /// The group this stream's page came from, taken at its first
    /// [cut](cut_at) and held until its last round. `None` while the stream is
    /// still free to have its page named by the round that finishes it, which
    /// is the common case — only a cut stream reserves ahead.
    page: Option<PageParity>,
    /// Whether this stream renders nothing at all: it is a layer covering no
    /// pixels, or nested inside one.
    suppressed: bool,
    /// The blurred backdrop pages the ops accumulated since this stream's last
    /// round sample — only ever the frame root's, since a backdrop is served
    /// nowhere else. They travel with the ops into whichever round emits them,
    /// which releases them.
    samples: Vec<BackdropSample>,
}

impl<'a> Stream<'a> {
    /// The frame root's stream, which renders into no page.
    fn root(nodes: &'a [Node]) -> Self {
        Self {
            owner: StreamOwner::Root,
            nodes,
            index: 0,
            depth: 0,
            ops: Vec::new(),
            emitted: false,
            page: None,
            suppressed: false,
            samples: Vec::new(),
        }
    }

    /// An inlined layer's stream, spliced into its parent's round at the depth
    /// its parent already occupies — an inlined layer takes no page, so it
    /// consumes no level (see the module header).
    fn inline(layer: &'a RecordedLayer, depth: usize, suppressed: bool) -> Self {
        Self {
            owner: StreamOwner::Inline,
            nodes: &layer.nodes,
            index: 0,
            depth,
            ops: Vec::new(),
            emitted: false,
            page: None,
            suppressed,
            samples: Vec::new(),
        }
    }

    /// An isolated layer's stream, one level deeper than its parent.
    fn isolated(id: u32, layer: &'a RecordedLayer, depth: usize, suppressed: bool) -> Self {
        Self {
            owner: StreamOwner::Isolated { id, layer },
            nodes: &layer.nodes,
            index: 0,
            depth,
            ops: Vec::new(),
            emitted: false,
            page: None,
            suppressed,
            samples: Vec::new(),
        }
    }

    /// A retained body's stream, at its parent's depth: the body renders into
    /// the store's page, not a pooled one, so it consumes no level of the
    /// chain its nested layers alternate along.
    fn retained(
        id: u32,
        layer: &'a RecordedLayer,
        bracket: u32,
        depth: usize,
        suppressed: bool,
    ) -> Self {
        Self {
            owner: StreamOwner::Retained { id, layer, bracket },
            nodes: &layer.nodes,
            index: 0,
            depth,
            ops: Vec::new(),
            emitted: false,
            page: None,
            suppressed,
            samples: Vec::new(),
        }
    }

    /// Whether this stream is one a page group is still owed to.
    ///
    /// True only for an isolated layer that will really render — a suppressed
    /// one renders nothing — and only until its first cut names its group.
    fn wants_a_page(&self) -> bool {
        matches!(self.owner, StreamOwner::Isolated { .. })
            && !self.suppressed
            && self.page.is_none()
    }
}

/// Whose stream a [`Stream`] walks, and so what finishing it produces.
enum StreamOwner<'a> {
    /// The frame's own surface. It occupies no page, which is what leaves both
    /// groups free for the layers directly under it.
    Root,
    /// A layer whose ops are spliced into its parent's round.
    Inline,
    /// A layer with a page and a round of its own.
    Isolated {
        /// The layer's index into
        /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
        id: u32,
        /// The recorded layer, for its bounds and its opacity.
        layer: &'a RecordedLayer,
    },
    /// A retained body, with rounds of its own into the store's page.
    Retained {
        /// The layer's index into
        /// [`CommandRecorder::layers`](vello_common::record::CommandRecorder::layers).
        id: u32,
        /// The recorded layer, for its bounds and its opacity.
        layer: &'a RecordedLayer,
        /// The retained bracket the page belongs to.
        bracket: u32,
    },
}

/// Which groups are holding a finished page.
///
/// A group holds one page at a time, so three booleans are this scheduler's
/// whole page allocator: acquiring is finding a group no round still needs,
/// releasing is the round that sampled a page ending. Nothing here allocates a
/// texture — the pool does that at execute time, keyed on the extent the round
/// carries.
///
/// The [ping-pong pair](PING_PONG_GROUPS) and the [spill page](Self::spill) are
/// kept apart deliberately. [`free`](Self::free) counts only the pair, so
/// [`make_room`]'s decisions — when to cut ahead of a shortage, when to cut at
/// one — are made against exactly the state they were tuned against and every
/// shape this scheduler already served schedules unchanged. The spill is only
/// ever reached through [`acquire_spill`](Self::acquire_spill), at the one site
/// that would otherwise refuse the frame.
#[derive(Debug, Default)]
struct LivePages {
    held: [bool; PING_PONG_GROUPS],
    /// Whether the one spill page is holding a page a later round composites.
    spill: bool,
}

impl LivePages {
    /// The ping-pong group to render into, preferring `preferred` and falling
    /// back to the other, or `None` when both are holding a page a later round
    /// still composites.
    ///
    /// The preference is what keeps a chain alternating exactly as its depths'
    /// parities say; the fallback is what lets a second sibling, which shares
    /// its predecessor's depth and so its preference, take the free group
    /// instead of overwriting the page beside it. Never answers the spill page:
    /// that is [`acquire_spill`](Self::acquire_spill)'s, so the fallback to a
    /// third page is a decision a caller makes rather than one this hides.
    fn acquire(&mut self, preferred: PageParity) -> Option<PageParity> {
        let parity = if !self.holds(preferred) {
            preferred
        } else if !self.holds(preferred.opposite()) {
            preferred.opposite()
        } else {
            return None;
        };

        self.set(parity, true);
        Some(parity)
    }

    /// The spill page, or `None` when it is already holding one.
    ///
    /// The whole of the bound: one page, so a second layer wanting it while it
    /// is held finds nothing and the frame is refused exactly as it was before
    /// the spill existed.
    fn acquire_spill(&mut self) -> Option<PageParity> {
        if self.spill {
            return None;
        }
        self.set(PageParity::Spill, true);
        Some(PageParity::Spill)
    }

    /// Hand a group's page back, once the round that sampled it has ended.
    fn release(&mut self, parity: PageParity) {
        self.set(parity, false);
    }

    /// How many of the ping-pong groups are holding no page.
    ///
    /// The spill page is deliberately not counted: it is a last resort rather
    /// than a group in the rotation, and counting it would make [`make_room`]
    /// stop cutting one shortage early — which would change how every already
    /// served shape schedules.
    fn free(&self) -> usize {
        self.held.iter().filter(|held| !**held).count()
    }

    /// Whether `parity`'s group is holding a page.
    fn holds(&self, parity: PageParity) -> bool {
        match parity {
            PageParity::Spill => self.spill,
            group => self.held.get(group.index()).copied().unwrap_or(false),
        }
    }

    fn set(&mut self, parity: PageParity, held: bool) {
        match parity {
            PageParity::Spill => self.spill = held,
            group => {
                if let Some(slot) = self.held.get_mut(group.index()) {
                    *slot = held;
                }
            }
        }
    }
}

/// How `layer` is served, refusing every property this scheduler cannot honour.
fn layer_role(id: u32, layer: &RecordedLayer) -> Result<LayerRole, EngineError> {
    if layer.props.blend_mode != BlendMode::default() {
        return Err(escalate(format!(
            "layer {id} composites with a non-default blend mode ({:?}), which has to read the \
             target it blends into",
            layer.props.blend_mode
        )));
    }
    if layer.props.mask.is_some() {
        return Err(escalate(format!(
            "layer {id} carries a layer mask, which needs an intermediate of its own"
        )));
    }
    if layer.props.clip_path.is_some() {
        return Err(escalate(format!(
            "layer {id} carries a layer clip path; frust lowers clips through the clip stack, not \
             through a layer"
        )));
    }

    let opacity = layer.props.opacity;
    if !opacity.is_finite() {
        return Err(escalate(format!(
            "layer {id} carries a non-finite opacity ({opacity})"
        )));
    }

    if opacity <= 0.0 {
        return Ok(LayerRole::Dropped);
    }

    match &layer.kind {
        RecordedLayerKind::Regular => Ok(if opacity >= 1.0 {
            LayerRole::Inline
        } else {
            LayerRole::Isolated
        }),
        // The two filters the engine renders are a Gaussian blur and a
        // shadow-only drop shadow; every other filter a recording can carry is
        // refused here, by name, and the frame is skipped. A filtered layer is
        // never inlined, whatever its opacity.
        kind => {
            // Recognised here rather than at the filter round, so a refusal
            // names what was found before any round has been emitted and a
            // filter shape this engine cannot render is refused wherever it is
            // recorded — including inside a layer whose contents cover nothing.
            served_filter(id, kind).map_err(escalate)?;
            Ok(LayerRole::Filtered)
        }
    }
}

/// Whether `ops` composites anything, and so whether emitting them as a round
/// would hand a page back to the pool.
fn holds_composite(ops: &[RoundOp]) -> bool {
    ops.iter().any(|op| matches!(op, RoundOp::Composite(_)))
}

/// The pages `ops` samples, each listed once.
fn released_pages(ops: &[RoundOp]) -> Vec<PageParity> {
    let mut pages: Vec<PageParity> = Vec::new();

    for op in ops {
        if let RoundOp::Composite(composite) = op
            && composite.retained.is_none()
            && !pages.contains(&composite.parity)
        {
            pages.push(composite.parity);
        }
    }

    pages
}

/// Why layer `id` is refused when every page it could have rendered into is
/// holding one a later round still composites.
///
/// Two reasons, because two different bounds are reached. `spills` is whether
/// the layer was offered the [spill page](PageParity::Spill) — every regular
/// layer is — and so whether what ran out was all three of this scheduler's
/// live pages or only the ping-pong pair a filter layer is held to.
fn out_of_pages(id: u32, spills: bool) -> String {
    if spills {
        format!(
            "layer {id} would need a fourth live intermediate page: both of the \
             {PING_PONG_GROUPS} groups the simple scheduler ping-pongs between and the one spill \
             page beside them are already holding a page a later round composites, and no open \
             round could be cut to hand one back. A nested chain of any depth fits, a fan of \
             siblings of any width fits, and so does a chain hanging off an isolated ancestor's \
             later child; what does not is a layer whose own round samples two live pages while a \
             third is still owed to a round above it"
        )
    } else {
        format!(
            "filter layer {id} would need a third live intermediate page for its contents: both \
             of the {PING_PONG_GROUPS} groups the simple scheduler ping-pongs between are already \
             holding a page a later round composites, and no open round could be cut to hand one \
             back. The spill page a regular layer falls back on is not offered here — a filter \
             layer's passes ping-pong between the two groups themselves, so it is held to them"
        )
    }
}

/// The escalation error carrying `reason`.
fn escalate(reason: impl Into<String>) -> EngineError {
    EngineError::SchedulerEscalation {
        reason: reason.into(),
    }
}