Read and bounded repair
- Zone state and KASI restore fromone stored fold; a restore holdsa request at most 500 ms.
+ Zone state, KASI and stationpages restore from stored folds;a restore holds a request at most500 ms.
@@ -5758,7 +5758,7 @@
Canonical reuse; conditional candidates
Canonical v2 raw/catalog/group controls remain the restore authority. Same selected original closure must replay exactly.
-
The collector stores zone-state daily, KASI capture stores kasi-fold and the image timer airkorea-forecast-hourly: folds of raw inputs, never assembled weather.
+
The collector stores zone-state daily, KASI capture stores kasi-fold, folding processes store station-pages per day and the image timer airkorea-forecast-hourly: folds of raw inputs, never assembled weather.
Per-grid optimal combinations, source plans, day locators, heads and packs remain deferred. Single EC2 runs only: the cold p95 gate is not qualified.
diff --git a/docs/architecture/diagrams/server2-kma-data-design.json b/docs/architecture/diagrams/server2-kma-data-design.json
index 84eaa0a35..163965e7d 100644
--- a/docs/architecture/diagrams/server2-kma-data-design.json
+++ b/docs/architecture/diagrams/server2-kma-data-design.json
@@ -240,7 +240,7 @@
"to": "s3",
"y": 1713,
"label": "Read and bounded repair",
- "note": "Zone state and KASI restore from one stored fold; a restore holds a request at most 500 ms."
+ "note": "Zone state, KASI and station pages restore from stored folds; a restore holds a request at most 500 ms."
},
{
"id": "reacquire",
@@ -285,7 +285,7 @@
"title": "Canonical reuse; conditional candidates",
"items": [
"Canonical v2 raw/catalog/group controls remain the restore authority. Same selected original closure must replay exactly.",
- "The collector stores zone-state daily, KASI capture stores kasi-fold and the image timer airkorea-forecast-hourly: folds of raw inputs, never assembled weather.",
+ "The collector stores zone-state daily, KASI capture stores kasi-fold, folding processes store station-pages per day and the image timer airkorea-forecast-hourly: folds of raw inputs, never assembled weather.",
"Per-grid optimal combinations, source plans, day locators, heads and packs remain deferred. Single EC2 runs only: the cold p95 gate is not qualified."
]
}
diff --git a/docs/architecture/server2-kma-data-design.md b/docs/architecture/server2-kma-data-design.md
index 3d356f444..c132de7d7 100644
--- a/docs/architecture/server2-kma-data-design.md
+++ b/docs/architecture/server2-kma-data-design.md
@@ -419,6 +419,7 @@ representations/v1/{source}/{kind}/{key_hash}/{day}/{manifest_sha256}.json
raw/v2/kma-warning/zone-state/{YYYY}/{MM}/{DD}/...
raw/v2/data-go-kr/kasi-fold/{YYYY}/{MM}/{DD}/...
raw/v2/airkorea-image/airkorea-forecast-hourly/{YYYY}/{MM}/{DD}/...
+raw/v2/kma-web/station-pages/{YYYY}/{MM}/{DD}/...
# Deferred discovery experiments; not initial serving dependencies:
index/v3/{source}/{kind}/{key_hash}/day/{YYYYMMDD}.json
@@ -505,7 +506,11 @@ capture pass. These hold provider items or legacy-shaped state, not assembled we
process falls back to the original objects when no usable fold exists. The AirKorea image hourly
forecast is the one fold requests read without that fallback: `airkorea-forecast-hourly`, the
legacy `ArpltnHourlyForecast` rows decoded from every stored image acquisition, is made by the
-process timer (below), and no request decodes the GIFs. The observed response also
+process timer (below), and no request decodes the GIFs. The station page product
+(`station-pages`, one per KST day) holds the parsed AWS hourly and city pages any folding process
+keeps; a restarted process reads it, compares it with the day catalogs and restores only the
+pages it lacks, falling back to the raw pages when it cannot be used
+([restart read](server2.md#s15b-2-restart-read-and-station-page-products-2026-11-07)). The observed response also
uses the release's land zones (`data/kma_forecast_land_zones.csv`) until a forecast-zone list is
stored, so its first answer can request mid temperatures. Measurements are in the
[speed evidence](../evidence/research/kma-speed-options-2026-11-04.md#result-of-the-change).
diff --git a/docs/architecture/server2.md b/docs/architecture/server2.md
index 2c99bcf03..4f7c70275 100644
--- a/docs/architecture/server2.md
+++ b/docs/architecture/server2.md
@@ -2270,6 +2270,89 @@ from `/gather` against the retired `KmaStnMinute`; the rules are ported but not
decision), and the deployed `ASOS_HISTORY_STATIONS` allowlist (every registry city station is a
candidate).
+### S15b-2 restart read and station page products (2026-11-07)
+
+The EC2 restart comparison of #2866 showed the first request after a restart without minute
+observations or past days: the load restored every stored hourly and city page of the window one
+day after another, parsed each again (about 1.0 to 1.4 s per day on a t3.small), and read the
+minute pages last through latest-first pages that read the descriptor of every minute group of the
+day. AK chose the day product design on 2026-11-07 ([intent](../../intent/issue-2614-kma-restart-observe.md),
+[spec](../../specs/issue-2614-kma-restart-observe.md), [plan](../../plans/issue-2614-kma-restart-observe.md)).
+
+**Targeted reads.** `CatalogStore::lookup_members` reads one catalog and then only the named
+members' groups (one descriptor and one raw body each); `lookup_acquisition` is its one-member
+case, and `ordered_candidates`/`lookup_page` are unchanged for the other cursor readers.
+`Reader::listed_acquisitions` lists a catalog's acquisitions from the control object alone and is
+shared with the image forecast. The minute catalogs of the readers' 40 minutes are listed once; of
+each minute the newest acquisition (completion, then raw digest) is restored, and when it is not
+complete or does not parse the next one of that minute (at most three per minute), at most four
+members per read.
+
+**Day product.** One product per KST day D: source `kma-web`, kind `station-pages`, the AWS page
+key, partition `YYYYMMDD`, period D/0000, fetched at its making instant, published as a one-member
+group. It is gzip JSON of the parsed inputs of the hourly fold, not its output: `schema`,
+`parseRevision` (bumped when page parsing or the reduction to registry facts changes), `made`,
+`day`, `excluded`, `knownIds`, `excludedIds`, `cityIds`, a `facts` table and `pages`. Per kind and hour of D it holds the
+newest page as the process keeps it (`kind`, `hour`, `fetchedAt`, `sha256`, `rawId`, `rows`).
+AWS rows equal to their registry facts are indices into `facts`; city rows are whole. `cityIds` is
+the city set the reduced rows are whole for (the intersection of the sets the day's pages were
+reduced with). Every listed group of the day is accounted for: its pages, `knownIds` (the raw
+identities of the other groups found complete, such as superseded acquisitions) and `excludedIds`
+(a member of each group S06 excluded; `excluded` is their number). Values keep every
+stored value (undefined, dates, negative zero, NaN) and key order. A reader refuses a product that
+expands past 16 MiB.
+
+Any process that folds writes products: after a fold that recomputed the hourly part a
+`Flight::Product` (never awaited by a request) stores the day whose made-of (sorted page raw
+identities, `cityIds`, excluded identities, `parseRevision`) differs from the last product this process read or stored.
+Only days read without failure whose every hour is in the window are stored, each at most once per
+ten minutes per process. Products are not cached on disk.
+
+**Restart read.** In parallel: the product, AWS hourly and city partition LISTs, the seed and the
+minute catalogs. Then for the due days, newest first and eight at a time, the day's page and
+product catalogs are listed (control objects only) and, the first time a process reads the day,
+its newest product (completion, then digest) is read and decoded. A product whose `cityIds` does
+not cover the current city set (seed and city pages) is not used. Product pages the catalogs do not
+list are dropped (S3 is truth). A day is then compared: every listed group that is neither held nor
+known complete (the excluded ones again, and those published since) is restored, four per catalog
+read; a complete one becomes known, an incomplete one excluded. Of each kind and hour the newest
+complete new page (latest fetch; of one instant the smaller digest, the page a read of every
+acquisition keeps) that is newer than the held one is parsed and held. An unreadable product, an
+older schema or parse revision, a grown city set, no product, or a new newest page that does not
+parse gives the day's raw read as before, which records every complete group as known and every
+excluded one. When a fold's city set has grown past the set an AWS page was reduced for (a city
+page restored after the products, or acquired since, names another station), `Flight::Expand`
+reads those pages in full again by their raw identity, four per catalog read, and keeps each read
+as it completes; no request waits for it. A page whose read was started waits a minute before it
+is tried again; pages not reached before the read's 30 s limit, and due retries, are started by the
+next fold or request without waiting for a new hourly page. The next fold reduces the pages read
+in full for the new set. `Flight::Load` ends after the products, today's
+and yesterday's comparisons, the seed and the four newest minute pages. `Flight::LoadRest`, which
+no request waits for, then compares the other days (each marked read and the pages dirty) and
+restores the rest of the minute window. Reloads every five minutes compare with the pages held.
+
+**S16.** Answers, the station page input identities and the response-first durability set are
+unchanged: a page restored from a product carries the raw identity and digest of the stored page.
+A day is partial while one of its listed groups is excluded, as a raw read with excluded groups is.
+Excluded groups are checked again at every read of the day (a restart, and each reload), so a
+repaired group completes the day and its product is stored again without it. A group published
+after a product, older than its pages, is unaccounted for and checked like any new group, so a
+restart from products counts a day complete exactly when a raw read would, given that a complete
+raw object is never deleted. `settled()` counts `LoadRest`, `Product`, `Expand` and a due product
+comparison as busy, and is false while a held AWS page is reduced for a city set that does not
+cover the last fold's (its past rows of a city station would be missing).
+Right after the first deployment the window's oldest day has no product (some of its hours were
+already out of the window) and is read from its raw pages after the first answer.
+
+Verification (local, S06 loopback S3 peer, no provider): targeted member GET counts, a restart over
+200 stored minute groups, the value codec round trip, a restart with unreachable providers answering
+168 past hourly pages and minute rows on its first request without raw page GETs, a later page
+restored alone and equal to a cold raw read, unusable products replaced by the raw pages and stored
+again, two processes' products, unsettled answers during the rest of the read, a day with an
+S06-excluded group kept partial, a repaired group completing its day (and recording summaries), a
+station joining the city set after the products (also with failing and with slow full reads), an
+incomplete newest minute acquisition, and the disk cache exclusion.
+
### S15b-2 s station registry seed (2026-10-15)
AK authorized a one-time read-only export of the legacy `kmastninfos` on 2026-10-15. It ran with
@@ -3040,6 +3123,13 @@ S16 also changes one S15b-2 behavior: a station-page day whose `aws-hourly` or `
is not listed now reads as empty. Before, the absent catalog's lookup answered "incomplete", so a
day without city pages never loaded its hourly pages from S3 and was read again on every load.
+Since the [restart read](#s15b-2-restart-read-and-station-page-products-2026-11-07) (2026-11-07) a
+station-page day can be read from its day product and the day catalogs instead of every raw page.
+The input identities are the same, a product carries the S06 exclusions of the read it was made
+from, and a station fold is not complete while the rest of a restart read or a product comparison
+is still running.
+The first answer after a start therefore records no summary; a later request does.
+
**Last-year lookup** (`summary::Lookup::last_year`, not wired to a route). The same date a year
earlier (Feb 29 → Feb 28), then ±1..±3, earlier first. At each date: the town's grid summary, then
the station's ASOS summary (when missing, one data.go.kr daily range `[base-3, base+3]` through the
diff --git a/docs/operations/server2-infrastructure.md b/docs/operations/server2-infrastructure.md
index 2284a2876..baddb1d5f 100644
--- a/docs/operations/server2-infrastructure.md
+++ b/docs/operations/server2-infrastructure.md
@@ -264,7 +264,9 @@ is read by the tests or measurement tool.
The cache holds verified immutable raw objects for `data-go-kr`, `kma-web`, `vc`
and the exact `airkorea-image` / `airkorea-forecast-image` pair. AirKorea's image
raws are included; the decoded hourly forecast (`airkorea-forecast-hourly`, made by
-the process timer and read from S3) and packs are **not** cached on disk.
+the process timer and read from S3), the KMA station page products
+(`kma-web` / `station-pages`, made by any process that folds the station pages) and packs
+are **not** cached on disk.
Mutable catalogs, absence proofs, state, budgets, push tokens, geocode and exact
request coordinates are excluded. Catalog/group publication checks still use S3.
Filename and namespace hashes do not replace the raw codec's identity validation.
diff --git a/intent/issue-2614-kma-restart-observe.md b/intent/issue-2614-kma-restart-observe.md
new file mode 100644
index 000000000..0bc4fe2d8
--- /dev/null
+++ b/intent/issue-2614-kma-restart-observe.md
@@ -0,0 +1,26 @@
+# Station observations after a restart
+
+AK requires that a request waits at most 3 s and answers from what was gathered by then. The EC2 restart comparison of #2866 showed that the first weather request after a restart misses the minute observations and the past days. The restart read restored every stored AWS hourly and city page of the 8-day window one day after another and parsed each page again: about 1.0 to 1.4 s per day on a t3.small, plus 4 to 5 s of sequential GETs. The minute pages came last, and every page call read the descriptor of every group in the day's minute catalog. The request stopped waiting at 2.5 s with neither. On 2026-11-07 AK chose the day product design ("A로 진행").
+
+## Outcomes
+
+- **MINUTE:** a restart lists the day's minute catalog once and restores only the newest acquisition of each minute within the readers' 40 minutes. The newest four come before the first answer and the rest follow.
+- **PRODUCT:** any process that folds the station pages stores one product per KST day holding the parsed inputs of the hourly fold, the newest page of each kind and hour, at most once per ten minutes per day and only when what the day is made of changed. No request waits for it.
+- **RESTART:** a restarted process reads each day's newest product and compares it with the day catalogs (S3 is truth). It restores only the pages the product lacks, parses only those, and answers past days and minute rows from its first request. Today and yesterday are compared before the answer, the other days after it.
+- **ACCOUNTED:** a repaired group completes its day after a restart, a station that joins the city set after the products has its AWS rows read in full again, and a minute whose newest acquisition is not complete is read from the next one, each as a raw read gives it.
+- **FALLBACK:** an unreadable product, an older schema or parse revision, or a city set that grew is replaced by the raw pages of that day, and the day's product is stored again. A page the catalog does not list is dropped.
+- **S16:** legacy answers, the station page input identities and the response-first durability set are unchanged. A day stays partial when its product's read had a group S06 excluded or a newer group is not complete, and answers stay unsettled until the rest of the read ends.
+- **PLACEMENT:** the placement gate and the artifact policy pass on the outgoing commits.
+
+## Decisions for AK
+
+This adds a stored object that other processes trust as a starting point, so it is not merged without AK:
+
+- Products are written to S3 under `kma-web` / `station-pages` beside the pages they are made of, never deleted, at most one per day and ten minutes per process. They are excluded from the disk cache, like the image forecast product.
+- A day product records every listed group of its day: the pages, the other complete groups and the S06-excluded ones. Each read checks the excluded groups and any group published since, so a restart counts a day complete exactly when a raw read would, as long as no complete raw object is ever deleted. While a group stays excluded, every read of its day spends one catalog read and the group's descriptor and raw GETs on it.
+- The first answer after a start is never complete for S16: the days past yesterday are compared after it, so a later request records the daily summaries. A fold that recomputes the hourly part also holds summaries back until its product comparison ends.
+- A process at hour H stores products only for days whose every hour is in the window. Right after the first deployment, the oldest day of the window therefore has no product and is read from its raw pages after the first answer.
+
+## Exclusions
+
+Not in this change: a real EC2 measurement of the new path, a product for the minute pages, the station ASOS state, and the collector's restart probing (KMA-SR-06 of #2862).
diff --git a/plans/issue-2614-kma-restart-observe.md b/plans/issue-2614-kma-restart-observe.md
new file mode 100644
index 000000000..5505191ff
--- /dev/null
+++ b/plans/issue-2614-kma-restart-observe.md
@@ -0,0 +1,21 @@
+# Plan for station observations after a restart
+
+Task declaration: `server2/config/tasks/KMA-RESTART-OBSERVE.json`. Follows `plans/issue-2614-kma-image-production.md` and the EC2 restart comparison of #2866.
+
+1. Read named members of one catalog (`CatalogStore::lookup_members`), and share the listing of a catalog's acquisitions (`Reader::listed_acquisitions`) between the image forecast and the observations.
+2. List the minute catalog once at a restart and restore only the newest acquisition of each minute within reach.
+3. Store the parsed station pages of a day as one product with a lossless value codec, and keep it out of the disk cache.
+4. Write the changed day products after a fold that recomputed the hourly part, never awaited by a request.
+5. Read the products at a restart, compare them with the day catalogs, restore only what they lack, and fall back to the raw pages when a product cannot be used. Finish today and yesterday before the answer and the rest after it, with S16 completeness kept.
+6. Update the architecture and operations documents.
+7. Open a draft PR and take it through CI and cross-provider review. AK decides on the merge.
+
+## Scenarios
+
+| User | Goal | Prerequisites | Actions | Expected | Failure path | Criteria |
+| --- | --- | --- | --- | --- | --- | --- |
+| App user | See current and past observations right after a deployment | Products of the window stored by the previous process | Request weather from the new process | Past hours and minute rows on the first answer, within 3 s | No product for a day: that day's raw pages follow after the answer | RESTART, MINUTE |
+| Operator | Restart without re-parsing every page | Products stored, pages unchanged | Start a new process | No raw page read for the product days; no product stored | A product unreadable or of an older revision: that day's raw pages are read and its product stored again | RESTART, FALLBACK |
+| Operator | Two hosts serve at once | Both fold the same pages | Both write their products; a third starts | The third reads the newest product of each day and answers as both | A page only one host has: the catalog comparison restores it | RESTART |
+| Summary recorder | Record only complete days | One day had a group S06 excluded | Restart from the products | The day stays partial; no summary recorded from it | The rest of the read still running: answers unsettled until it ends | S16 |
+| Operator | Keep S3 writes bounded | A long-running process | Requests every few minutes | A day's product only when its pages changed, at most once per ten minutes | A failed write is tried again ten minutes later | PRODUCT |
diff --git a/server2/config/tasks/KMA-RESTART-OBSERVE.json b/server2/config/tasks/KMA-RESTART-OBSERVE.json
new file mode 100644
index 000000000..c67dcc12d
--- /dev/null
+++ b/server2/config/tasks/KMA-RESTART-OBSERVE.json
@@ -0,0 +1,68 @@
+{
+ "task": "KMA-RESTART-OBSERVE",
+ "issue": "https://github.com/WizardFactory/TodayWeather/issues/2614",
+ "scope": "Observed KMA path: a restarted process restores station observations without restoring every stored page. Targeted member reads and one listing of the minute catalog; a per-day product of the parsed station pages written by any process that folds, read at restart and compared with the day catalogs, with the remaining days restored after the first answer. No deployment or production mutation.",
+ "server2_paths": [
+ "server2/config/tasks/KMA-RESTART-OBSERVE.json",
+ "server2/src/cache/disk.rs",
+ "server2/src/domestic/kma/source/forecast_input.rs",
+ "server2/src/domestic/kma/source/mod.rs",
+ "server2/src/domestic/kma/source/observe.rs",
+ "server2/src/domestic/kma/source/observe_product.rs",
+ "server2/src/domestic/kma/source/records.rs",
+ "server2/src/storage/catalog.rs",
+ "server2/src/storage/codec.rs",
+ "server2/src/storage/response_first.rs",
+ "server2/tests/cache.rs",
+ "server2/tests/domestic.rs",
+ "server2/tests/domestic/full_response.rs",
+ "server2/tests/domestic/observe.rs",
+ "server2/tests/domestic/observe_product.rs",
+ "server2/tests/domestic/summary_s3.rs",
+ "server2/tests/response_first.rs",
+ "server2/tests/storage/catalog_wire.rs",
+ "server2/tests/storage/codec.rs"
+ ],
+ "outside": [
+ {
+ "path": "intent/issue-2614-kma-restart-observe.md",
+ "category": "documentation",
+ "reason": "Task intent."
+ },
+ {
+ "path": "specs/issue-2614-kma-restart-observe.md",
+ "category": "documentation",
+ "reason": "Acceptance."
+ },
+ {
+ "path": "plans/issue-2614-kma-restart-observe.md",
+ "category": "documentation",
+ "reason": "Plan."
+ },
+ {
+ "path": "docs/architecture/server2.md",
+ "category": "documentation",
+ "reason": "The observation restart read, the station page product and its S16 note."
+ },
+ {
+ "path": "docs/architecture/server2-kma-data-design.md",
+ "category": "documentation",
+ "reason": "The station page product among the server-written folds."
+ },
+ {
+ "path": "docs/operations/server2-infrastructure.md",
+ "category": "documentation",
+ "reason": "The station page product is not cached on disk."
+ },
+ {
+ "path": "docs/architecture/diagrams/server2-kma-data-design.json",
+ "category": "documentation",
+ "reason": "Archify source: station pages restore from a stored fold."
+ },
+ {
+ "path": "docs/architecture/diagrams/server2-kma-data-design.html",
+ "category": "documentation",
+ "reason": "Archify output regenerated from the source."
+ }
+ ]
+}
diff --git a/server2/src/cache/disk.rs b/server2/src/cache/disk.rs
index 315019dd3..d63147d43 100644
--- a/server2/src/cache/disk.rs
+++ b/server2/src/cache/disk.rs
@@ -90,9 +90,13 @@ struct Header {
fn failed() -> io::Error {
io::Error::other("disposable cache unavailable")
}
+/// Provider raw records; the products processes make from them (the AirKorea hourly forecast,
+/// the KMA station page products) are read from S3.
fn eligible(id: &RecordId) -> bool {
id.validate().is_ok()
- && (matches!(id.source.as_str(), "data-go-kr" | "kma-web" | "vc")
+ && (matches!(id.source.as_str(), "data-go-kr" | "vc")
+ || (id.source == "kma-web"
+ && id.kind != crate::domestic::kma::source::observe_product::KIND)
|| (id.source == "airkorea-image" && id.kind == "airkorea-forecast-image"))
}
fn token(name: &str) -> bool {
diff --git a/server2/src/domestic/kma/source/forecast_input.rs b/server2/src/domestic/kma/source/forecast_input.rs
index 34cb97a13..086d67cc9 100644
--- a/server2/src/domestic/kma/source/forecast_input.rs
+++ b/server2/src/domestic/kma/source/forecast_input.rs
@@ -40,8 +40,8 @@ use crate::gateway::BoxFuture;
use crate::history::IndexListing;
use crate::resolver::{OwnedConfig, OwnedFlights};
use crate::storage::{
- Catalog, CatalogId, CatalogLimits, CatalogStore, Envelope, GroupDeclaration, GroupMember,
- PageOutcome, Pagination, RawRecord,
+ CatalogId, CatalogLimits, CatalogStore, Envelope, GroupDeclaration, GroupMember, PageOutcome,
+ Pagination, RawRecord,
};
use std::collections::BTreeMap;
use std::sync::Arc;
@@ -289,44 +289,19 @@ fn image_catalog(kind: &str, day: &str) -> Result {
partition: day.into(),
})
}
-/// The acquisitions one day catalog lists: (completion instant, fetch group, a member envelope),
-/// from the catalog control object alone (no record body is read).
-async fn listed_acquisitions(
- transport: &T,
+/// The acquisitions one day catalog lists ([`Reader::listed_acquisitions`]).
+async fn listed_acquisitions(
+ reader: &Reader,
id: &CatalogId,
deadline: Instant,
) -> Result, String> {
- let limits = CatalogLimits::default();
- let object = match tokio::time::timeout_at(
- deadline,
- transport.get_control(&id.key(), limits.control_bytes),
- )
- .await
- {
- Err(_) => return Err("image catalog deadline".into()),
- Ok(Err(crate::storage::Error::NotFound)) => return Ok(Vec::new()),
- Ok(Err(e)) => return Err(format!("image catalog: {e}")),
- Ok(Ok(object)) => object,
- };
- let catalog = Catalog::from_bytes(&object.body, id, &limits)
- .map_err(|e| format!("image catalog: {e}"))?;
- // One acquisition per fetch group; its completion is its members' latest fetch instant.
- let mut groups: BTreeMap = BTreeMap::new();
- for e in catalog.entries {
- let group = match &e.fetch_group {
- Some(g) => g.descriptor_key().map_err(|e| e.to_string())?,
- None => e.identity.raw_sha256.clone(),
- };
- let fetched = e.identity.fetched_at_ms;
- groups
- .entry(group)
- .and_modify(|(f, _)| *f = (*f).max(fetched))
- .or_insert((fetched, e));
- }
- Ok(groups
- .into_iter()
- .map(|(group, (fetched, e))| (fetched, group, e))
- .collect())
+ reader
+ .listed_acquisitions(id, deadline)
+ .await
+ .map_err(|e| match e {
+ crate::storage::Error::Timeout => "image catalog deadline".to_owned(),
+ e => format!("image catalog: {e}"),
+ })
}
impl ImageForecasts {
@@ -769,9 +744,7 @@ async fn listed(
let mut out = Vec::new();
for day in window(cut).into_iter().filter(|d| listed.contains(d)) {
let id = image_catalog(image::KIND, &day)?;
- for (fetched, group, member) in
- listed_acquisitions(&shared.reader.transport, &id, deadline).await?
- {
+ for (fetched, group, member) in listed_acquisitions(&shared.reader, &id, deadline).await? {
if fetched as f64 <= cut {
out.push(((fetched, day.clone(), group), id.clone(), member));
}
@@ -929,7 +902,7 @@ async fn newest_product(
.map_err(|e| format!("product listing: {e}"))?;
for day in days.iter().rev().filter(|d| **d >= oldest) {
let id = image_catalog(PRODUCT_KIND, day)?;
- let newest = listed_acquisitions(&shared.reader.transport, &id, deadline)
+ let newest = listed_acquisitions(&shared.reader, &id, deadline)
.await?
.into_iter()
.max_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1)));
diff --git a/server2/src/domestic/kma/source/mod.rs b/server2/src/domestic/kma/source/mod.rs
index 1fd3258e1..0c25cea72 100644
--- a/server2/src/domestic/kma/source/mod.rs
+++ b/server2/src/domestic/kma/source/mod.rs
@@ -43,6 +43,7 @@ pub mod live;
pub mod mfds;
mod mid;
pub mod observe;
+pub mod observe_product;
pub mod pages;
mod records;
mod reference;
diff --git a/server2/src/domestic/kma/source/observe.rs b/server2/src/domestic/kma/source/observe.rs
index 4da289329..6c47e081b 100644
--- a/server2/src/domestic/kma/source/observe.rs
+++ b/server2/src/domestic/kma/source/observe.rs
@@ -19,6 +19,7 @@
//! the next request. Nothing runs without a request.
use super::asos::{AsosFetched, AsosStore};
use super::doc::{Doc, Val};
+use super::observe_product::{self as product, DayProduct, MadeOf, StoredPage};
use super::pages::{merge_aws_city, parse_aws, parse_city};
use super::records::Reader;
use super::representative::{self, Verdict};
@@ -31,7 +32,7 @@ use crate::domestic::publication::Publication;
use crate::domestic::{Acquired, DomesticError};
use crate::gateway::{BoxFuture, OrderedJson};
use crate::history::{IndexListing, schedule};
-use crate::storage::{CatalogStore, ProviderKey, RawRecord};
+use crate::storage::{CatalogId, CatalogStore, ProviderKey, RawRecord};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::{Arc, Mutex};
use std::time::Duration;
@@ -78,6 +79,17 @@ const ONGOING_MS: f64 = 2. * 3_600_000.;
const MEMO_KEEP: Duration = Duration::from_secs(7 * 3600);
/// Minute pages kept in memory (the readers look back 20 minutes).
const MINUTE_KEEP_MS: f64 = 40. * 60_000.;
+/// Minute pages restored before the load ends (the newest; the rest follow in
+/// [`Flight::LoadRest`]), and the most members of one targeted read.
+const MINUTE_FIRST: usize = 4;
+/// A day's product is stored at most once per this interval by one process.
+const PRODUCT_EVERY: Duration = Duration::from_secs(600);
+/// A reduced AWS hourly page whose full read was tried is read again after this long (an S3
+/// read, retried as a failed ASOS S3 read is).
+const EXPAND_RETRY: Duration = Duration::from_secs(60);
+/// Acquisitions of one minute tried, newest first, when a newer one is not complete or does not
+/// parse.
+const MINUTE_CANDIDATES: usize = 3;
/// Legacy minute collection cadence.
const MINUTE_CADENCE: i64 = 2;
/// S3 day partitions read again after this long (other processes may have published).
@@ -170,6 +182,12 @@ fn page_ids(ids: &PageIds) -> Arc> {
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Flight {
Load,
+ /// The rest of a load after the answer: the other days' missing pages and minute pages.
+ LoadRest,
+ /// Store the day products whose pages changed.
+ Product,
+ /// Read AWS hourly pages in full again whose reduction misses a city station.
+ Expand,
Fold,
Geocode,
Minute,
@@ -201,8 +219,19 @@ struct Pages {
ids: PageIds,
minute: BTreeMap>>,
days: HashMap,
- /// Days whose latest S3 read failed (S16 completeness; a later success clears them).
+ /// Days whose latest S3 read failed or was partial (S16 completeness; a later complete read
+ /// clears them).
failed_days: std::collections::BTreeSet,
+ /// Days whose latest S3 read failed (no product is stored for them).
+ unreadable: BTreeSet,
+ /// Raw identities of the listed groups of each read day found complete beyond the held
+ /// pages (superseded ones; S16 accounts for every listed group).
+ known: BTreeMap>,
+ /// Raw identities of a member of each listed group of a read day S06 excluded (S16; checked
+ /// again at every read of the day).
+ excluded: BTreeMap>,
+ /// AWS hourly pages kept reduced to registry facts outside this city set.
+ reduced: BTreeMap>>,
minute_read: Option,
/// Pages changed since the last fold.
dirty: bool,
@@ -234,6 +263,14 @@ struct State {
registry_id: Option<(String, f64)>,
/// The hourly part of the last fold (registry, hourly rows, national rows).
hourly_fold: Option,
+ /// What the last product of each day this process read or stored is made of.
+ products: HashMap,
+ /// When this process last tried to store each day's product.
+ product_attempts: HashMap,
+ /// A fold recomputed the hourly part since the products were last compared.
+ product_due: bool,
+ /// When this process last read each reduced AWS hourly page in full again.
+ expand_attempts: HashMap,
}
#[derive(Clone)]
struct HourlyFold {
@@ -242,6 +279,8 @@ struct HourlyFold {
national: Arc>>,
unresolved: Vec,
pages: Arc>,
+ /// The city set the hourly rows were kept in full for.
+ cities: Arc>,
}
pub struct Observations {
@@ -458,6 +497,11 @@ impl Observations {
if let Some(rx) = self.fold_if_dirty(now_ms) {
Self::wait(vec![rx], until).await;
}
+ // A product comparison a finishing one missed, and a due full read of reduced pages.
+ if self.state().product_due && self.flight(&Flight::Product).is_none() {
+ self.want_products(now_ms);
+ }
+ self.expand_due();
// Before the first fold the carried seed alone is the registry (no pages yet).
Some(self.state().folded.clone().unwrap_or_else(seed_only))
}
@@ -471,8 +515,11 @@ impl Observations {
rx.clone()
})
}
- /// Read hourly/city day partitions of the 8-day window and the latest minute pages from S3
- /// (at start and every [`RELOAD`]).
+ /// Read the hourly/city pages of the 8-day window and the latest minute pages from S3 (at
+ /// start and every [`RELOAD`]). The load ends once the request can answer: every day's
+ /// stored product (the parsed pages of that day), the pages of today and yesterday the
+ /// products lack, the seed and the newest minute pages. [`Flight::LoadRest`] then restores the
+ /// other days' missing pages and the rest of the minute window; no request waits for it.
fn load(self: &Arc, now_ms: i64) -> Option> {
let stale = {
let s = self.state();
@@ -489,180 +536,607 @@ impl Observations {
let deadline = Instant::now() + READ_DEADLINE;
let first = schedule::from_ms(now_ms - HOURLY_WINDOW_HOURS * 3_600_000 - 3_600_000);
let today = schedule::from_ms(now_ms);
- let mut days = Vec::new();
- let mut d = Publication::new(first.date(), 0);
- while let Some(day) = d {
- if day.date() > today.date() {
- break;
- }
- days.push(day.date());
- d = day.add_hours(24);
- }
- // Published day catalogs of both page kinds (S16): a day whose catalog is not listed has
- // nothing published and reads as empty, as S14/S15b listings treat absent catalogs. A
- // failed listing keeps every day to its catalog lookup.
+ let days = window_days(now_ms);
+ // Published day catalogs of both page kinds and of the products (S16): a day whose catalog
+ // is not listed has nothing published and reads as empty, as S14/S15b listings treat
+ // absent catalogs. A failed listing keeps every day to its catalog read.
let months = super::records::months(schedule::to_ms(first) as f64, now_ms as f64);
let (aws_k, city_k) = (aws_key(), city_key());
- let (listed_hourly, listed_city) = tokio::join!(
+ let (listed_products, listed_hourly, listed_city, seed, mut minutes) = tokio::join!(
+ self.reader
+ .partitions("kma-web", product::KIND, &aws_k, &months, deadline),
self.reader
.partitions("kma-web", "aws-hourly", &aws_k, &months, deadline),
self.reader
.partitions("kma-web", "city-obs", &city_k, &months, deadline),
+ self.read_seed(now_ms, deadline),
+ self.minute_entries(now_ms, deadline),
);
+ self.apply_seed(seed);
let listed = |l: &Result, crate::storage::Error>, p: &str| {
l.as_ref().map_or(true, |l| l.iter().any(|x| x == p))
};
- for day in days {
- let fresh = {
- let s = self.state();
- s.pages.days.get(&day).is_some_and(|at| {
- day < today.date() && at.elapsed() < Duration::from_secs(86_400)
- || at.elapsed() < RELOAD
+ // Due days newest first: their catalogs and products, LOOKUPS at a time.
+ let due: Vec<(u32, [bool; 3])> = {
+ let s = self.state();
+ days.iter()
+ .rev()
+ .filter(|day| {
+ !s.pages.days.get(day).is_some_and(|at| {
+ **day < today.date() && at.elapsed() < Duration::from_secs(86_400)
+ || at.elapsed() < RELOAD
+ })
+ })
+ .map(|day| {
+ let p = format!("{day:08}");
+ (
+ *day,
+ [
+ listed(&listed_hourly, &p),
+ listed(&listed_city, &p),
+ listed(&listed_products, &p),
+ ],
+ )
})
+ .collect()
+ };
+ use futures_util::StreamExt;
+ let mut reads: Vec = futures_util::stream::iter(due)
+ .map(|(day, want)| self.list_day(day, want, deadline))
+ .buffered(super::records::LOOKUPS)
+ .collect()
+ .await;
+ self.apply_products(&mut reads);
+ // Today and yesterday are compared now; the other days after the answer.
+ let yesterday = today.add_hours(-24).map_or(today.date(), Publication::date);
+ let (now, rest): (Vec, Vec) =
+ reads.drain(..).partition(|r| r.day >= yesterday);
+ crate::history::join_all(
+ now.into_iter()
+ .map(|r| self.complete_day(r, deadline))
+ .collect(),
+ )
+ .await;
+ let floor = schedule::from_ms(now_ms - MINUTE_KEEP_MS as i64);
+ let later = minutes.split_off(minutes.len().min(MINUTE_FIRST));
+ self.restore_minutes(minutes, floor, deadline).await;
+ {
+ let mut s = self.state();
+ s.pages.minute.retain(|k, _| *k > floor);
+ s.pages.minute_read = Some(std::time::Instant::now());
+ s.pages.dirty = true;
+ let oldest = first.date();
+ s.pages.days.retain(|d, _| *d >= oldest);
+ s.pages.failed_days.retain(|d| *d >= oldest);
+ s.pages.unreadable.retain(|d| *d >= oldest);
+ s.pages.excluded.retain(|d, _| *d >= oldest);
+ s.pages.known.retain(|d, _| *d >= oldest);
+ s.products.retain(|d, _| *d >= oldest);
+ s.product_attempts.retain(|d, _| *d >= oldest);
+ }
+ if !rest.is_empty() || !later.is_empty() {
+ let _ = self.start(Flight::LoadRest, Duration::ZERO, move |this| {
+ Box::pin(async move {
+ for day in rest {
+ this.complete_day(day, Instant::now() + READ_DEADLINE).await;
+ }
+ this.restore_minutes(later, floor, Instant::now() + READ_DEADLINE)
+ .await;
+ })
+ });
+ }
+ }
+ /// Use the seed read (S16: a failed read leaves the registry unproven; the previous seed stays
+ /// in use).
+ fn apply_seed(&self, seed: Result<(Option>, bool), ()>) {
+ let mut s = self.state();
+ s.seed_failed = !matches!(seed, Ok((_, false)));
+ if let Ok((Some(seed), _)) = seed
+ && s.seed
+ .as_ref()
+ .is_none_or(|old| old.len() != seed.len() || **old != *seed)
+ {
+ s.seed = Some(Arc::new(seed));
+ s.pages.dirty = true;
+ s.pages.hourly_dirty = true;
+ }
+ }
+ /// List one day's page catalogs (control objects only) and, the first time this process
+ /// reads the day, read and decode its newest product.
+ async fn list_day(&self, day: u32, want: [bool; 3], deadline: Instant) -> DayRead {
+ let partition = format!("{day:08}");
+ let known = self.state().pages.days.contains_key(&day);
+ let list = |kind: &'static str, key: ProviderKey, wanted: bool| {
+ let partition = partition.clone();
+ async move {
+ if !wanted {
+ return Ok(None);
+ }
+ let id = super::records::catalog("kma-web", kind, &key, &partition)?;
+ let entries = self.reader.listed_acquisitions(&id, deadline).await?;
+ Ok::<_, crate::storage::Error>(Some((id, entries)))
+ }
+ };
+ let (hourly, city, products) = tokio::join!(
+ list("aws-hourly", aws_key(), want[0]),
+ list("city-obs", city_key(), want[1]),
+ list(product::KIND, aws_key(), want[2] && !known),
+ );
+ let listed = match (hourly, city) {
+ (Ok(hourly), Ok(city)) => {
+ let mut pages = Vec::new();
+ for (kind, listed) in [("aws-hourly", hourly), ("city-obs", city)] {
+ let Some((id, entries)) = listed else {
+ continue;
+ };
+ for (_, _, e) in entries {
+ if let Some(page) = ListedPage::new(kind, &id, e) {
+ pages.push(page);
+ }
+ }
+ }
+ Some(pages)
+ }
+ _ => None,
+ };
+ let newest = match (&listed, products) {
+ (Some(_), Ok(Some((id, entries)))) => entries
+ .into_iter()
+ .max_by(|a, b| {
+ (a.0, &a.2.identity.raw_sha256).cmp(&(b.0, &b.2.identity.raw_sha256))
+ })
+ .map(|(_, _, e)| (id, e.identity)),
+ _ => None,
+ };
+ let product = match newest {
+ Some((id, member)) => self.read_product(&id, member, day, deadline).await,
+ None => None,
+ };
+ DayRead {
+ day,
+ listed,
+ product,
+ base: None,
+ }
+ }
+ /// One stored product of `day`, decoded; None when it cannot be read or is not a product of
+ /// this schema and parse revision.
+ async fn read_product(
+ &self,
+ id: &CatalogId,
+ member: crate::storage::RecordId,
+ day: u32,
+ deadline: Instant,
+ ) -> Option {
+ let read = self
+ .reader
+ .members(id, std::slice::from_ref(&member), deadline)
+ .await
+ .ok()
+ .flatten()?;
+ let bytes = read.first()?.first()?.bytes.clone();
+ crate::observability::spawn_blocking(move || DayProduct::decode(&bytes, day))
+ .await
+ .ok()
+ .flatten()
+ }
+ /// Hold the pages of each read product that its day catalogs list (S3 is truth), unless the
+ /// product's city set does not cover the current one (its reduced rows may lack a city
+ /// station's fields): such a day reads its raw pages. A held product's day is then compared
+ /// with its catalogs; a day this process read before is compared with what it holds.
+ fn apply_products(&self, reads: &mut [DayRead]) {
+ let mut s = self.state();
+ let cities = {
+ let seed = s
+ .seed
+ .clone()
+ .unwrap_or_else(|| Arc::new(super::seed::embedded().docs.clone()));
+ let held = s.pages.city.values().map(|rows| &**rows);
+ let stored = reads
+ .iter()
+ .filter_map(|r| r.product.as_ref())
+ .flat_map(|p| p.pages.iter())
+ .filter(|page| page.kind == "city-obs")
+ .map(|page| &*page.rows);
+ city_ids(Some(&seed), held.chain(stored))
+ };
+ for r in reads.iter_mut() {
+ if s.pages.days.contains_key(&r.day) {
+ r.base = Some(Accounted {
+ known: s.pages.known.get(&r.day).cloned().unwrap_or_default(),
+ excluded: s.pages.excluded.get(&r.day).cloned().unwrap_or_default(),
+ });
+ continue;
+ }
+ let (Some(listed), Some(stored)) = (&r.listed, r.product.take()) else {
+ continue;
};
- if fresh {
+ if !stored.city_ids.is_superset(&cities) {
continue;
}
- let partition = format!("{day:08}");
- let (aws, city_k) = (aws_key(), city_key());
- let (want_hourly, want_city) = (
- listed(&listed_hourly, &partition),
- listed(&listed_city, &partition),
- );
- let (hourly, city) = tokio::join!(
- async {
- if want_hourly {
- self.reader
- .raw_counted("kma-web", "aws-hourly", &aws, &partition, deadline)
- .await
+ s.products.insert(r.day, stored.made_of());
+ let reduced = Arc::new(stored.city_ids.clone());
+ let mut stored = stored;
+ for page in std::mem::take(&mut stored.pages) {
+ if !listed.iter().any(|l| l.raw_id == page.raw_id) {
+ continue;
+ }
+ let reduced = (page.kind == "aws-hourly").then(|| reduced.clone());
+ let held = (
+ page.hour,
+ page.rows,
+ page.fetched_at_ms,
+ page.sha256,
+ page.raw_id,
+ );
+ hold(&mut s.pages, page.kind, held, reduced);
+ }
+ r.base = Some(Accounted {
+ known: stored.known,
+ excluded: stored.excluded,
+ });
+ s.pages.dirty = true;
+ s.pages.hourly_dirty = true;
+ }
+ }
+ /// Bring one day up to its catalogs. With pages held for it (a product, or an earlier read),
+ /// only the listed groups not accounted for are restored: the excluded ones again and the new
+ /// ones, and a new page newer than the held one of its kind and hour is parsed and held.
+ /// Otherwise, or when such a page does not parse, every page of the day is read as before
+ /// (the newest complete page of each hour).
+ async fn complete_day(&self, r: DayRead, deadline: Instant) {
+ let DayRead {
+ day, listed, base, ..
+ } = r;
+ let Some(listed) = listed else {
+ self.day_failed(day);
+ return;
+ };
+ if let Some(base) = base {
+ match self.compare_day(day, &listed, base, deadline).await {
+ Ok(Some((changed, accounted))) => return self.day_read(day, accounted, changed),
+ Ok(None) => {}
+ Err(()) => return self.day_failed(day),
+ }
+ }
+ self.raw_day(day, &listed, deadline).await;
+ }
+ /// [`Self::complete_day`] with pages held. Ok(None): a new newest page does not parse.
+ async fn compare_day(
+ &self,
+ day: u32,
+ listed: &[ListedPage],
+ base: Accounted,
+ deadline: Instant,
+ ) -> Result