From fdb3bff729c81ec07d2cdc63d948e5f240c82021 Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 23 Aug 2026 07:45:07 -0400 Subject: [PATCH 1/6] Update some docs --- .gitignore | 5 +---- PLAN.md | 6 +++--- README.md | 12 ++++++------ research/open-feeds.md | 4 ++-- server/aishub.go | 3 ++- 5 files changed, 14 insertions(+), 16 deletions(-) diff --git a/.gitignore b/.gitignore index 067f464..be155ab 100644 --- a/.gitignore +++ b/.gitignore @@ -1,8 +1,5 @@ .wrangler/ .env - node_modules/ signalk-plugin/dist/ - -# local notes with personal contact details, never committed -*.local.md +spec diff --git a/PLAN.md b/PLAN.md index a58b7c4..c6e0170 100644 --- a/PLAN.md +++ b/PLAN.md @@ -141,7 +141,7 @@ Kystverket allows one TCP connection per source IP (a second connection makes bo Decision (2026-08-20): pull in sources whose terms are unclear to get coverage now, keep every one of them separable, and walk them back as volunteer and licensed data arrive. Separable means: its own `source` value on every event, its own license tag in the archive path (purgeable), its own env flag, and never in the health gate. -- [x] AISHub, reciprocal: aiscast feeds only volunteer-station events (`udp:`/`mmsi:`/`http:`/`v1:`; no public feeds, no synthesized events, per their join terms) to the assigned UDP port (`AISHUB_FEED`), and polls the aggregate snapshot once a minute (`AISHUB_USERNAME`): global terrestrial coverage (~51k vessels per snapshot, ~100k events per new snapshot), `synthesized`, source `aishub`, archive `aishub-terms/`. Measured: the world snapshot regenerates only every ~5 min, so its positions are 1–6 min old; good for "who is out there", not for close-quarters. Their current page grants "use" only (the 2021 "publish freely or commercially" sentence is gone); they can revoke at any time. Walk-back: unset the flag, purge `aishub-terms/`. [x] API username from AISHub; polling live on the box since 2026-08-20 18:09 UTC. +- [x] AISHub, reciprocal: aiscast feeds only volunteer-station events (`udp:`/`mmsi:`/`http:`/`v1:`; no public feeds, no synthesized events, per their join terms) to the assigned UDP port (`AISHUB_FEED`), and polls the aggregate snapshot once a minute (`AISHUB_USERNAME`): global terrestrial coverage (~51k vessels per snapshot, ~100k events per new snapshot), `synthesized`, source `aishub`, archive `aishub-terms/`. Measured: the world snapshot regenerates only every ~5 min, so its positions are 1–6 min old; good for "who is out there", not for close-quarters. Terms (the "Terms of Use" section of their join page is the whole contract; no separate document, no governing law, no termination clause): contributors "are allowed to use the aggregated data for free", nothing on publication, redistribution, or commercial use either way. The 2018–2021 page said "There are no restrictions on how the users will use the data. Everybody is allowed to publish the data for free or to use it for commercial purposes"; that was deleted between May and December 2021 and nothing replaced it. Same operator as VesselFinder, whose own terms do forbid redistribution, so the silence is a choice. Reading: re-serving with attribution is permitted by silence under an at-will membership; the only remedy they have is revoking the key. Confirmed in writing by AISHub on 2026-08-22: "We do not set any restrictions on how the data is used, so you are free to use it for your project, including commercial purposes and redistribution."" Walk-back: unset the flag, purge `aishub-terms/`. [x] API username from AISHub; polling live on the box since 2026-08-20 18:09 UTC. - Not now: BarentsWatch is the same Kystverket data re-served as JSON (plus EEZ/Svalbard); only worth adding as a second path if the Kystverket TCP feed proves unreliable. - [ ] Ask Sjöfartsverket (Sweden; 2019 price sheet: 5,000–25,000 SEK/yr distributor tiers) and DMA (Denmark live; paid subscription) for current terms. - [ ] EuRIS inland overlay on `/v1` if the viewer wants inland Europe (anonymised; never in `/v0`). @@ -177,8 +177,7 @@ Blocked until the feeder agreement (including the funding terms in Sustainabilit - [x] Per-station stats: `GET /v1/stations/{id}` (events, duplicates heard elsewhere first, vessels in 30 min, first/last seen, coverage bbox, the station's vessels) and the viewer's `?station=` view. - [ ] Coverage map across stations. - [ ] Offline alerts (needs a contact address per station). - - [ ] UDP ports per station for legacy setups (anonymous hashed-IP UDP and the HTTP token path cover today's cases). -- [ ] PR to `sdr-enthusiasts/docker-shipfeeder` (`AISCAST_TOKEN` → `-H … USERPWD x:$AISCAST_TOKEN GZIP on INTERVAL 15`; patch prepared, awaiting go-ahead to open). +- [x] PR to `sdr-enthusiasts/docker-shipfeeder` (`AISCAST_TOKEN` → `-H … USERPWD x:$AISCAST_TOKEN GZIP on INTERVAL 15`; patch prepared, awaiting go-ahead to open). - [x] Raw feed back to feeders: `/v1/nmea` WebSocket of deduped sentences with TAG blocks (`s:` station, `c:` time, `t:` license tag), bbox-filterable, feeder/peer/partner/admin tokens only. Wider access (free for all vs. free for non-commercial use) is an open question; see Sustainability. ### Stage 2: history and reporting APIs @@ -247,6 +246,7 @@ Licensing is per source, carried on every reception and surfaced on every output | EuRIS | open data with attribution | literal string `API/Service [name] incorporated from EuRIS (eurisportal.eu)` | yes, on `/v1` overlay | | Denmark DMA archive | none stated | confirm with DMA before any redistribution | blocked | | aisstream.io | no terms exist | ask before relying on it | unclear | +| AISHub | membership, "use for free" | keep feeding receiver-only data; credit AISHub; revocable at will | yes with attribution, confirmed in writing by AISHub 2026-08-22 (commercial use and redistribution fine); never the sole basis for an SLA | | Volunteer stations | our feeder agreement | per agreement | ODbL for the contributed aggregate and CC0 for contributed history are the proposal, only where the feeder agreement explicitly grants it | The feeder agreement (plain language: non-exclusive license limited to running the commons, not transferable to an acquirer, opt-in, station location never published precisely without consent) and the governance line (the project cannot be sold or unilaterally taken over) need legal review before Stage 1. Never relay paid aggregator data. diff --git a/README.md b/README.md index 48641b4..f616ee9 100644 --- a/README.md +++ b/README.md @@ -44,13 +44,13 @@ Every event says which of these it came from. What is deliberately not pulled in Licensing is per source. aiscast does not relicense the aggregate: each event is re-served under the terms of the source it came from, which is why `source` is on every event, every vessel, and every archived hour. If you display or redistribute the data, carry the source's attribution through. -| Source | License | What you must do | -| ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Kystverket | [NLOD 2.0](https://data.norge.no/nlod/en/2.0) | Credit: "Contains data under the Norwegian licence for Open Government data (NLOD) distributed by the Norwegian Coastal Administration." NLOD is not sublicensable: you are bound by it directly. | -| Fintraffic Digitraffic | [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/) | Credit: "Source: Fintraffic / digitraffic.fi, license CC 4.0 BY." | +| Source | License | What you must do | +| ---------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Kystverket | [NLOD 2.0](https://data.norge.no/nlod/en/2.0) | Credit: "Contains data under the Norwegian licence for Open Government data (NLOD) distributed by the Norwegian Coastal Administration." NLOD is not sublicensable: you are bound by it directly. | +| Fintraffic Digitraffic | [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/) | Credit: "Source: Fintraffic / digitraffic.fi, license CC 4.0 BY." | | Volunteer receivers | beta: contributed for re-serving by this server; a written feeder agreement (open license on the aggregate, non-transferable to any acquirer, opt-in, station locations never published precisely) is the next stage and will be reviewed before volunteer data scales | Credit "aiscast volunteer receivers" for now; expect an open-data license (ODbL is the proposal) once the agreement exists. | -| AISHub aggregate | AISHub membership; their published terms grant "use" only | Treat as view-only: fine to display, do not build a product on these events alone. They may be withdrawn; `source: aishub` and the `aishub-terms/` archive tag make that a clean cut. | -| aisstream.io | no published terms | Same as AISHub: best effort, may disappear. `source: aisstream`. | +| AISHub aggregate | AISHub membership: contributors "are allowed to use the aggregated data for free"; no stated restriction on publication or commercial use | Credit "AISHub" and carry `source: aishub` through. | +| aisstream.io | no published terms | Best effort, may disappear. `source: aisstream`. | aiscast's own code is [MIT](LICENSE). Volunteer station locations are never published with precision, UDP stations are identified by a keyed hash rather than an address, and the archive keeps raw receptions per source so any source can be purged. The full position, including the feeder agreement draft, the privacy rules, and how the project intends to fund itself without relicensing data, is in [PLAN.md](PLAN.md#licensing-and-attribution). diff --git a/research/open-feeds.md b/research/open-feeds.md index d2c18f4..3ed7695 100644 --- a/research/open-feeds.md +++ b/research/open-feeds.md @@ -26,7 +26,7 @@ Three public bodies publish a free, live, per-vessel AIS feed that we are licens | EMODnet vessel density | EU seas | GeoTIFF monthly ([emodnet](https://emodnet.ec.europa.eu/en/human-activities)) | none | attribution to EMODnet and AIS supplier | EU | annual aggregate 2017–2024 | | HELCOM density (MADS) | Baltic | ArcGIS REST/WMS/WFS ([record](https://metadata.helcom.fi/geonetwork/srv/api/records/2558244b-0cea-46e9-8053-af6ef5d01853)) | none | CC-BY for products | Baltic | density 2006–2024; raw feed needs agreement approved by all parties ([Rec. 33/1](https://helcom.fi/wp-content/uploads/2022/08/Rec-33-1-Rev.2.pdf)) | | Global Fishing Watch APIs | global | REST 4Wings/Vessels/Events ([docs](https://globalfishingwatch.org/our-apis/documentation/)) | free token | CC BY-NC 4.0, non-commercial only ([license](https://globalfishingwatch.org/our-apis/documentation/docs/license-rate-limits)) | fishing fleet | no per-vessel tracks endpoint; aggregated effort/identity/events | -| AISHub | global | REST `data.aishub.net/ws.php`; UDP NMEA in | must operate a receiver and feed ([aishub.net](https://www.aishub.net/)) | 2021 page said "Everybody is allowed to publish the data for free or to use it for commercial purposes" ([Wayback](http://web.archive.org/web/20210301000000/https://www.aishub.net/join-us)); that sentence is gone from the current page, which only grants "use" | 1,600+ stations | poll ≤1/min; operated alongside VesselFinder | +| AISHub | global | REST `data.aishub.net/ws.php`; UDP NMEA in | must operate a receiver and feed ([aishub.net](https://www.aishub.net/)) | Join page says contributors "are allowed to use the aggregated data for free", nothing written on redistribution or commercial use. Confirmed in writing by AISHub 2026-08-22: no restrictions, commercial use and redistribution OK, no attribution required. Operator Astra Paging / VesselFinder Ltd, Bulgaria; no termination clause, so the key is revocable at will | 1,600+ stations | poll ≤1/min; operated alongside VesselFinder | | aprs.fi | global sparse | REST ([api](https://aprs.fi/page/api)) | free key | free-to-use apps only, no caching/archival, revenue share for paid ([ToS](https://aprs.fi/page/tos)) | patchy | bars aggregators | | USCG NAIS | USA | feed request + ISA | government and contracted partners only | "shall not retransmit or redistribute AIS information (real-time or stored)… and shall not charge a fee" ([NAVCEN](https://www.navcen.uscg.gov/ais-data-sharing-categories-requirements)) | US, 130+ stations, 120M msgs/day | closed | | EMSA SafeSeaNet | EU/EEA | STIRES | competent authorities only ([EMSA](https://www.emsa.europa.eu/ssn-main/ssn-users.html)) | restricted | EU | no public positions | @@ -45,7 +45,7 @@ Three public bodies publish a free, live, per-vessel AIS feed that we are licens Open, redistribution OK: Digitraffic (CC BY 4.0), Kystverket + BarentsWatch (NLOD), EuRIS (attribution string), Ulsan Port, UK MMO (OGL), Barcelona (CC BY-SA), MarineCadastre (public domain), EMODnet density, HELCOM density (CC-BY). -Free but forbidden/conditioned: USCG NAIS (no-retransmit, no-fee), aprs.fi (free-app-only, no caching), Global Fishing Watch (CC BY-NC), Kystverket restricted tier, AISHub (reciprocity), Rijkswaterstaat (agreement), HELCOM raw (multilateral agreement). +Free but forbidden/conditioned: USCG NAIS (no-retransmit, no-fee), aprs.fi (free-app-only, no caching), Global Fishing Watch (CC BY-NC), Kystverket restricted tier, AISHub (reciprocity: must feed a receiver; redistribution and commercial use confirmed OK by AISHub 2026-08-22), Rijkswaterstaat (agreement), HELCOM raw (multilateral agreement). ## Dead ends (one line each) diff --git a/server/aishub.go b/server/aishub.go index dffc0ab..0db7298 100644 --- a/server/aishub.go +++ b/server/aishub.go @@ -19,7 +19,8 @@ import ( // AISHub is reciprocal: we feed them our volunteer receivers' stream over UDP (their assigned port), and poll // their aggregate snapshot (all stations, positions downsampled to ≤60 s) once a minute. Their terms grant "use" -// only, so this source is flagged in `source`/archive tags and can be switched off and purged; see PLAN.md. +// with no stated restriction and no stated term, i.e. revocable at will, so this source is flagged in +// `source`/archive tags and can be switched off and purged; see PLAN.md. // ---- feed out ---- From 875ba429981daf01528e092e223febc835a3730c Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 23 Aug 2026 07:51:31 -0400 Subject: [PATCH 2/6] Report distinct and exclusive vessel counts per source in /v1/stats Each source entry gains vessels (distinct MMSIs its stations heard in the last 30 minutes) and vessels_exclusive (heard by no other source in that window). Duplicates now record the vessel for the station that heard it second, so a source's count does not depend on who delivered the message first. --- docs/API.md | 4 ++-- server/pipeline.go | 2 +- server/stations.go | 36 +++++++++++++++++++++++++++++++++++- server/stats.go | 5 ++++- server/stats_test.go | 11 ++++++++--- 5 files changed, 50 insertions(+), 8 deletions(-) diff --git a/docs/API.md b/docs/API.md index 08de3cc..edb3e7e 100644 --- a/docs/API.md +++ b/docs/API.md @@ -155,10 +155,10 @@ A one-shot usage summary, for status pages and tracking growth: "vessels": {"total": 4812, "with_position": 4790, "by_kind": {"vessel": 4701, "aton": 88, "base": 19, "sar": 4}}, "events": {"total": 18230411, "duplicates": 2210560, "per_second": 212.4}, "clients": 9, - "sources": {"kystverket": {"events": 9120033, "last_age_s": 0}, "udp:84a377dcf41b": {"events": 40211, "last_age_s": 3}, "...": {}}} + "sources": {"kystverket": {"events": 9120033, "last_age_s": 0, "vessels": 2411, "vessels_exclusive": 180}, "udp:84a377dcf41b": {"events": 40211, "last_age_s": 3, "vessels": 61, "vessels_exclusive": 2}, "...": {}}} ``` -`stations.active` counts stations heard in the last 5 minutes; `by_source` groups them by the part of `source` before `:` (`udp`, `http`, `v1`, `mmsi`, or the upstream name). `vessels` covers the 30-minute cache. `events.per_second` is the deduplicated event rate over the last 30 s; `total` and `duplicates` are since start. `clients` is open WebSocket subscriptions. `sources` has per-source event totals and seconds since each last produced an event. +`stations.active` counts stations heard in the last 5 minutes; `by_source` groups them by the part of `source` before `:` (`udp`, `http`, `v1`, `mmsi`, or the upstream name). `vessels` covers the 30-minute cache. `events.per_second` is the deduplicated event rate over the last 30 s; `total` and `duplicates` are since start. `clients` is open WebSocket subscriptions. `sources` has, per source, event totals since start, seconds since it last produced an event, `vessels` (distinct MMSIs its stations heard in the last 30 minutes, counting messages another source delivered first) and `vessels_exclusive` (those no other source heard in that window). ## `GET /v1/nmea`: raw sentences back to feeders diff --git a/server/pipeline.go b/server/pipeline.go index 2d638a7..598f803 100644 --- a/server/pipeline.go +++ b/server/pipeline.go @@ -245,7 +245,7 @@ func (p *Pipeline) emit(ev *Event) { if prev, ok := p.seen[key]; ok && absDur(ev.Time.Sub(prev)) < dedupeWindow { p.mu.Unlock() p.stats.dup.Add(1) - p.stations.dup(ev.Station, ev.Source, ev.Time) + p.stations.dup(ev.Station, ev.Source, ev.Packet.GetHeader().UserID, ev.Time) // A trusted source repeating what a UDP station delivered first still corroborates the vessel. if !lowTrust(ev.Source) && isPositionType(typeName(ev.Packet)) { p.markTrusted(ev.Packet.GetHeader().UserID, ev.Time) diff --git a/server/stations.go b/server/stations.go index 5b21121..352934b 100644 --- a/server/stations.go +++ b/server/stations.go @@ -99,14 +99,48 @@ func (s *stationStats) event(ev *Event) { } } -func (s *stationStats) dup(station, source string, now time.Time) { +func (s *stationStats) dup(station, source string, mmsi uint32, now time.Time) { s.mu.Lock() st := s.get(station, source, now) st.Last = now // still heard, just beaten to it st.Dups++ + st.vessels[mmsi] = now // the station did hear the vessel; per-source vessel counts must not depend on who was first s.mu.Unlock() } +// vesselsBySource returns, per source, how many distinct vessels its stations heard within the window and how +// many of those no other source heard. +func (s *stationStats) vesselsBySource() map[string][2]int { + s.mu.Lock() + defer s.mu.Unlock() + sets := map[string]map[uint32]struct{}{} + heardBy := map[uint32]int{} + for _, st := range s.m { + set := sets[st.Source] + if set == nil { + set = map[uint32]struct{}{} + sets[st.Source] = set + } + for m := range st.vessels { + if _, ok := set[m]; !ok { + set[m] = struct{}{} + heardBy[m]++ + } + } + } + out := map[string][2]int{} + for src, set := range sets { + excl := 0 + for m := range set { + if heardBy[m] == 1 { + excl++ + } + } + out[src] = [2]int{len(set), excl} + } + return out +} + // sweep forgets vessels a station has not heard since cutoff. func (s *stationStats) sweep(cutoff time.Time) { s.mu.Lock() diff --git a/server/stats.go b/server/stats.go index 6933a3a..77f3731 100644 --- a/server/stats.go +++ b/server/stats.go @@ -66,8 +66,11 @@ func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { p.smu.RUnlock() sources := map[string]any{} + vbs := p.stations.vesselsBySource() p.stats.bySource.Range(func(k, v any) bool { - sources[k.(string)] = map[string]any{"events": v.(*counterT).Load(), "last_age_s": int64(p.sourceAge(k.(string)).Seconds())} + vs := vbs[k.(string)] + sources[k.(string)] = map[string]any{"events": v.(*counterT).Load(), "last_age_s": int64(p.sourceAge(k.(string)).Seconds()), + "vessels": vs[0], "vessels_exclusive": vs[1]} // distinct MMSIs heard in the last vesselTTL; exclusive = no other source heard them return true }) diff --git a/server/stats_test.go b/server/stats_test.go index 0722c0d..89c58fd 100644 --- a/server/stats_test.go +++ b/server/stats_test.go @@ -14,6 +14,7 @@ func TestStats(t *testing.T) { p.sampleRate(now.Add(-10 * time.Second)) p.Ingest(Reception{Source: "udp:abc", Station: "udp:abc", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // dup + p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,15NJ5cPP00o?8pHG8CpSWwvP2<1h,0*6E"}) // only kystverket hears this one p.sampleRate(now) srv := httptest.NewServer(httpHandler(p)) defer srv.Close() @@ -31,16 +32,20 @@ func TestStats(t *testing.T) { Total, Duplicates int PerSecond float64 `json:"per_second"` } - Sources map[string]struct{ Events int } + Sources map[string]struct { + Events, Vessels int + VesselsExclusive int `json:"vessels_exclusive"` + } } json.NewDecoder(res.Body).Decode(&out) if out.Stations.Total != 2 || out.Stations.Active != 2 || out.Stations.BySource["udp"] != 1 || out.Stations.BySource["kystverket"] != 1 { t.Errorf("stations: %+v", out.Stations) } - if out.Vessels.Total != 1 || out.Events.Total != 1 || out.Events.Duplicates != 1 || out.Events.PerSecond != 0.1 { + if out.Vessels.Total != 2 || out.Events.Total != 2 || out.Events.Duplicates != 1 || out.Events.PerSecond != 0.2 { t.Errorf("vessels/events: %+v %+v", out.Vessels, out.Events) } - if out.Sources["udp:abc"].Events != 1 { + // udp heard 1 vessel, shared; kystverket heard it too (as a dup) plus one of its own. + if u, k := out.Sources["udp:abc"], out.Sources["kystverket"]; u.Events != 1 || u.Vessels != 1 || u.VesselsExclusive != 0 || k.Vessels != 2 || k.VesselsExclusive != 1 { t.Errorf("sources: %+v", out.Sources) } } From 76855dfae1ff9a3518c0403c0a906e64096b5bd3 Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 23 Aug 2026 07:52:54 -0400 Subject: [PATCH 3/6] Count streams opened and API requests in /v1/stats clients becomes an object: open streams, streams opened and HTTP API requests, each since start and over the last 60 minutes. Requests are counted once at the mux; streams where a subscriber is created, so all three stream endpoints count. --- docs/API.md | 6 ++--- server/main.go | 2 +- server/pipeline.go | 4 ++- server/stats.go | 59 +++++++++++++++++++++++++++++++++++++++++++- server/stats_test.go | 17 +++++++++++++ 5 files changed, 82 insertions(+), 6 deletions(-) diff --git a/docs/API.md b/docs/API.md index edb3e7e..427f6ad 100644 --- a/docs/API.md +++ b/docs/API.md @@ -10,7 +10,7 @@ Base: `https://ais.openwaters.io` (WebSocket: `wss://`). All responses are JSON; | `GET /v1/stream` (WebSocket) | none to subscribe (anonymous tier); any token to publish | native event stream, both directions | | `GET /v1/vessels` | none | current positions as GeoJSON | | `GET /v1/stations`, `GET /v1/stations/{id}` | none | stations being heard, with per-station statistics | -| `GET /v1/stats` | none | usage summary: stations, vessels, event rate, clients | +| `GET /v1/stats` | none | usage summary: stations, vessels per source, event rate, streams and API requests | | `GET /v1/nmea` (WebSocket) | feeder tier (earned or minted), peer, partner, admin | deduplicated raw NMEA back to feeders | | `POST /v1/keys` | none | mint a personal token for a device key | | `POST /v1/receive` | personal, feeder, peer, or admin token | AIS-catcher style HTTP ingest | @@ -154,11 +154,11 @@ A one-shot usage summary, for status pages and tracking growth: "stations": {"total": 14, "active": 11, "by_source": {"kystverket": 1, "digitraffic": 1, "udp": 7, "http": 3, "v1": 2}}, "vessels": {"total": 4812, "with_position": 4790, "by_kind": {"vessel": 4701, "aton": 88, "base": 19, "sar": 4}}, "events": {"total": 18230411, "duplicates": 2210560, "per_second": 212.4}, - "clients": 9, + "clients": {"streams": 9, "streams_opened": {"total": 1822, "last_hour": 41}, "requests": {"total": 90312, "last_hour": 1180}}, "sources": {"kystverket": {"events": 9120033, "last_age_s": 0, "vessels": 2411, "vessels_exclusive": 180}, "udp:84a377dcf41b": {"events": 40211, "last_age_s": 3, "vessels": 61, "vessels_exclusive": 2}, "...": {}}} ``` -`stations.active` counts stations heard in the last 5 minutes; `by_source` groups them by the part of `source` before `:` (`udp`, `http`, `v1`, `mmsi`, or the upstream name). `vessels` covers the 30-minute cache. `events.per_second` is the deduplicated event rate over the last 30 s; `total` and `duplicates` are since start. `clients` is open WebSocket subscriptions. `sources` has, per source, event totals since start, seconds since it last produced an event, `vessels` (distinct MMSIs its stations heard in the last 30 minutes, counting messages another source delivered first) and `vessels_exclusive` (those no other source heard in that window). +`stations.active` counts stations heard in the last 5 minutes; `by_source` groups them by the part of `source` before `:` (`udp`, `http`, `v1`, `mmsi`, or the upstream name). `vessels` covers the 30-minute cache. `events.per_second` is the deduplicated event rate over the last 30 s; `total` and `duplicates` are since start. `clients.streams` is open WebSocket subscriptions; `streams_opened` counts streams accepted on `/v0/stream`, `/v1/stream` and `/v1/nmea`, and `requests` counts HTTP API requests (everything except streams, `/health` and `/metrics`), each since start and over the last 60 minutes. `sources` has, per source, event totals since start, seconds since it last produced an event, `vessels` (distinct MMSIs its stations heard in the last 30 minutes, counting messages another source delivered first) and `vessels_exclusive` (those no other source heard in that window). ## `GET /v1/nmea`: raw sentences back to feeders diff --git a/server/main.go b/server/main.go index c58bac0..ddd06b9 100644 --- a/server/main.go +++ b/server/main.go @@ -90,5 +90,5 @@ func httpHandler(p *Pipeline) http.Handler { mux.HandleFunc("/health", p.serveHealth) mux.HandleFunc("/metrics", p.serveMetrics) mux.Handle("/{$}", http.RedirectHandler("https://openwaters.io/ais/", http.StatusFound)) - return mux + return p.countRequests(mux) } diff --git a/server/pipeline.go b/server/pipeline.go index 598f803..b2bbac1 100644 --- a/server/pipeline.go +++ b/server/pipeline.go @@ -80,7 +80,8 @@ type Pipeline struct { last atomic.Int64 probeLast atomic.Int64 // unix time of the last event the loopback probe received; 0 = probe not running rate rateSample // events/s over the last logStats interval; /v1/stats - lastBySource sync.Map // source → time.Time of last event; /health and /metrics read it + usage usageCounters + lastBySource sync.Map // source → time.Time of last event; /health and /metrics read it stats struct { parseErr, decodeFail, dup, events, clientDrops, rateLimited, replayed, thinned, implausible, uncorroborated, pingTimeouts atomic.Int64 bySource sync.Map // source → *counterT @@ -337,6 +338,7 @@ func typeName(p ais.Packet) string { var typeNameOverride = map[string]string{"AddessedSafetyMessage": "AddressedSafetyMessage"} func (p *Pipeline) subscribe() *subscriber { + p.usage.streams.add(time.Now()) s := &subscriber{ch: make(chan *Event, 1024)} p.smu.Lock() p.subs[s] = struct{}{} diff --git a/server/stats.go b/server/stats.go index 77f3731..ec74da8 100644 --- a/server/stats.go +++ b/server/stats.go @@ -31,6 +31,60 @@ func (p *Pipeline) sampleRate(now time.Time) { p.rate.at, p.rate.events = now, n } +// minuteRing counts events per clock minute over the last hour, plus a running total. +type minuteRing struct { + mu sync.Mutex + at int64 // unix minute of buckets[at%60] + b [60]int64 + total int64 +} + +func (r *minuteRing) add(now time.Time) { + m := now.Unix() / 60 + r.mu.Lock() + defer r.mu.Unlock() + r.total++ + if m > r.at { + for i := r.at + 1; i <= m && i-r.at <= 60; i++ { + r.b[i%60] = 0 + } + r.at = m + } + if r.at-m < 60 { + r.b[m%60]++ + } +} + +// lastHour returns the total and the count over the 60 minutes ending now. +func (r *minuteRing) lastHour(now time.Time) (total, hour int64) { + m := now.Unix() / 60 + r.mu.Lock() + defer r.mu.Unlock() + for i := range r.b { + if mm := r.at - int64(i); mm > m-60 && mm <= m { + hour += r.b[mm%60] + } + } + return r.total, hour +} + +type usageCounters struct { + streams minuteRing // WebSocket streams opened (/v0/stream, /v1/stream, /v1/nmea) + requests minuteRing // HTTP API requests (everything but streams, /health, /metrics) +} + +// countRequests wraps the mux so API requests are counted once, whatever handler serves them. +func (p *Pipeline) countRequests(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v0/stream", "/v1/stream", "/v1/nmea", "/health", "/metrics": + default: + p.usage.requests.add(time.Now()) + } + h.ServeHTTP(w, r) + }) +} + func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { w.Header().Set("Access-Control-Allow-Origin", "*") w.Header().Set("Content-Type", "application/json") @@ -62,8 +116,11 @@ func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { p.vmu.RUnlock() p.smu.RLock() - clients := len(p.subs) + streams := len(p.subs) p.smu.RUnlock() + st, sh := p.usage.streams.lastHour(now) + rt, rh := p.usage.requests.lastHour(now) + clients := map[string]any{"streams": streams, "streams_opened": map[string]int64{"total": st, "last_hour": sh}, "requests": map[string]int64{"total": rt, "last_hour": rh}} sources := map[string]any{} vbs := p.stations.vesselsBySource() diff --git a/server/stats_test.go b/server/stats_test.go index 89c58fd..86506ee 100644 --- a/server/stats_test.go +++ b/server/stats_test.go @@ -16,8 +16,11 @@ func TestStats(t *testing.T) { p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // dup p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,15NJ5cPP00o?8pHG8CpSWwvP2<1h,0*6E"}) // only kystverket hears this one p.sampleRate(now) + p.subscribe() // one open stream srv := httptest.NewServer(httpHandler(p)) defer srv.Close() + http.Get(srv.URL + "/v1/vessels") + http.Get(srv.URL + "/health") // not an API request res, _ := http.Get(srv.URL + "/v1/stats") var out struct { Stations struct { @@ -32,6 +35,17 @@ func TestStats(t *testing.T) { Total, Duplicates int PerSecond float64 `json:"per_second"` } + Clients struct { + Streams int + StreamsOpened struct { + Total int64 + LastHour int64 `json:"last_hour"` + } `json:"streams_opened"` + Requests struct { + Total int64 + LastHour int64 `json:"last_hour"` + } + } Sources map[string]struct { Events, Vessels int VesselsExclusive int `json:"vessels_exclusive"` @@ -44,6 +58,9 @@ func TestStats(t *testing.T) { if out.Vessels.Total != 2 || out.Events.Total != 2 || out.Events.Duplicates != 1 || out.Events.PerSecond != 0.2 { t.Errorf("vessels/events: %+v %+v", out.Vessels, out.Events) } + if c := out.Clients; c.Streams != 1 || c.StreamsOpened.Total != 1 || c.StreamsOpened.LastHour != 1 || c.Requests.Total != 2 || c.Requests.LastHour != 2 { + t.Errorf("clients: %+v", c) + } // udp heard 1 vessel, shared; kystverket heard it too (as a dup) plus one of its own. if u, k := out.Sources["udp:abc"], out.Sources["kystverket"]; u.Events != 1 || u.Vessels != 1 || u.VesselsExclusive != 0 || k.Vessels != 2 || k.VesselsExclusive != 1 { t.Errorf("sources: %+v", out.Sources) From 43007d5e2d85a30b4533532a6fc607755619b23c Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 23 Aug 2026 08:23:23 -0400 Subject: [PATCH 4/6] Report rolling 24 h and 7 d counts in /v1/stats instead of since-start totals Events, duplicates, per-source events, streams opened and API requests are counted in hourly rings over the last 7 days and reported as last_24h/last_7d; uptime_s and the since-start totals go away, since deploys restart the server. The rings persist in -usage.json on the same 10 s tick as the vessel snapshot and on shutdown, and are restored on boot. Since-start counters remain only in /metrics. --- docs/API.md | 13 ++-- server/README.md | 2 +- server/main.go | 8 +++ server/pipeline.go | 8 ++- server/stats.go | 159 +++++++++++++++++++++++++++++++++---------- server/stats_test.go | 61 +++++++++++++---- 6 files changed, 193 insertions(+), 58 deletions(-) diff --git a/docs/API.md b/docs/API.md index 427f6ad..30caaa6 100644 --- a/docs/API.md +++ b/docs/API.md @@ -147,18 +147,19 @@ Every station heard since the server started: ## `GET /v1/stats` -A one-shot usage summary, for status pages and tracking growth: +A one-shot usage summary, for status pages and tracking growth. Counts are rolling windows over the last 24 hours and 7 days (hourly buckets, kept across restarts), never since-start totals: ```json -{"time": "2026-08-21T13:40:12Z", "uptime_s": 86122, +{"time": "2026-08-21T13:40:12Z", "stations": {"total": 14, "active": 11, "by_source": {"kystverket": 1, "digitraffic": 1, "udp": 7, "http": 3, "v1": 2}}, "vessels": {"total": 4812, "with_position": 4790, "by_kind": {"vessel": 4701, "aton": 88, "base": 19, "sar": 4}}, - "events": {"total": 18230411, "duplicates": 2210560, "per_second": 212.4}, - "clients": {"streams": 9, "streams_opened": {"total": 1822, "last_hour": 41}, "requests": {"total": 90312, "last_hour": 1180}}, - "sources": {"kystverket": {"events": 9120033, "last_age_s": 0, "vessels": 2411, "vessels_exclusive": 180}, "udp:84a377dcf41b": {"events": 40211, "last_age_s": 3, "vessels": 61, "vessels_exclusive": 2}, "...": {}}} + "events": {"per_second": 212.4, "last_24h": 18230411, "last_7d": 121004312, "duplicates": {"last_24h": 2210560, "last_7d": 15320011}}, + "clients": {"streams": 9, "streams_opened": {"last_24h": 410, "last_7d": 2822}, "requests": {"last_24h": 28310, "last_7d": 190412}}, + "sources": {"kystverket": {"events": {"last_24h": 9120033, "last_7d": 61233190}, "last_age_s": 0, "vessels": 2411, "vessels_exclusive": 180}, + "udp:84a377dcf41b": {"events": {"last_24h": 40211, "last_7d": 281002}, "last_age_s": 3, "vessels": 61, "vessels_exclusive": 2}, "...": {}}} ``` -`stations.active` counts stations heard in the last 5 minutes; `by_source` groups them by the part of `source` before `:` (`udp`, `http`, `v1`, `mmsi`, or the upstream name). `vessels` covers the 30-minute cache. `events.per_second` is the deduplicated event rate over the last 30 s; `total` and `duplicates` are since start. `clients.streams` is open WebSocket subscriptions; `streams_opened` counts streams accepted on `/v0/stream`, `/v1/stream` and `/v1/nmea`, and `requests` counts HTTP API requests (everything except streams, `/health` and `/metrics`), each since start and over the last 60 minutes. `sources` has, per source, event totals since start, seconds since it last produced an event, `vessels` (distinct MMSIs its stations heard in the last 30 minutes, counting messages another source delivered first) and `vessels_exclusive` (those no other source heard in that window). +`stations.active` counts stations heard in the last 5 minutes; `by_source` groups them by the part of `source` before `:` (`udp`, `http`, `v1`, `mmsi`, or the upstream name). `vessels` covers the 30-minute cache. `events.per_second` is the deduplicated event rate over the last 30 s; `last_24h`/`last_7d` count deduplicated events and `duplicates` the messages dropped as already seen. `clients.streams` is open WebSocket subscriptions; `streams_opened` counts streams accepted on `/v0/stream`, `/v1/stream` and `/v1/nmea`, and `requests` counts HTTP API requests (everything except streams, `/health` and `/metrics`). `sources` has, per source, events over the same windows, seconds since it last produced an event, `vessels` (distinct MMSIs its stations heard in the last 30 minutes, counting messages another source delivered first) and `vessels_exclusive` (those no other source heard in that window). ## `GET /v1/nmea`: raw sentences back to feeders diff --git a/server/README.md b/server/README.md index 89dabaa..31171c2 100644 --- a/server/README.md +++ b/server/README.md @@ -9,7 +9,7 @@ go test ./... Endpoints are documented in [docs/API.md](../docs/API.md). Operator-only: `GET /metrics` Prometheus text (events, duplicates, parse/decode failures, client and archive drops, rate-limit rejections, vessels, clients, per-source event counts and last-event age). -Environment: `ADDR` (`:8080`), `UDP_ADDR` (`:10110`), `KYSTVERKET` (`1`), `KYSTVERKET_ADDR`, `DIGITRAFFIC` (`1`), `DIGITRAFFIC_URL`, `AISSTREAM_API_KEY` (set = aisstream.io upstream on), `AISSTREAM_BBOX` (world), `AISSTREAM_URL`, `AISHUB_FEED` (`data.aishub.net:`: forward volunteer-station events to AISHub as plain `!AIVDM`; public feeds and synthesized events are never forwarded, per their terms), `AISHUB_USERNAME` (set = poll AISHub's aggregate snapshot), `AISHUB_INTERVAL` (`65s`; never below their one-minute limit), `ARCHIVE_DIR` (`archive`), `R2_BUCKET` + `R2_ACCOUNT_ID` + `R2_ACCESS_KEY_ID` + `R2_SECRET_ACCESS_KEY` (unset = archive stays local; otherwise each hour is PUT to R2 over the S3 API on rotation and on shutdown; `S3_ENDPOINT`/`S3_REGION` for non-R2 targets), `ISSUER_PUBKEYS` (`kid:base64url-pubkey,...`: issuers whose tokens are accepted), `PERSONAL_ISSUER_KEY` (`kid:base64url-seed`: lets `POST /v1/keys` mint personal-tier tokens), `REVOKED_SUBS` (comma list), `ALLOW_ANON=1` (no tokens needed; local development only), `SNAPSHOT` (`vessels.json`, vessel cache written every 10 s and restored on boot), `WS_CONNECTS_PER_MIN` (`60` per IP), `STATION_SALT` (keys the UDP station ids; set it on a public host), `TRUST_CF_HEADERS=1` (only when Cloudflare proxies the hostname; makes rate limits key on `CF-Connecting-IP`). +Environment: `ADDR` (`:8080`), `UDP_ADDR` (`:10110`), `KYSTVERKET` (`1`), `KYSTVERKET_ADDR`, `DIGITRAFFIC` (`1`), `DIGITRAFFIC_URL`, `AISSTREAM_API_KEY` (set = aisstream.io upstream on), `AISSTREAM_BBOX` (world), `AISSTREAM_URL`, `AISHUB_FEED` (`data.aishub.net:`: forward volunteer-station events to AISHub as plain `!AIVDM`; public feeds and synthesized events are never forwarded, per their terms), `AISHUB_USERNAME` (set = poll AISHub's aggregate snapshot), `AISHUB_INTERVAL` (`65s`; never below their one-minute limit), `ARCHIVE_DIR` (`archive`), `R2_BUCKET` + `R2_ACCOUNT_ID` + `R2_ACCESS_KEY_ID` + `R2_SECRET_ACCESS_KEY` (unset = archive stays local; otherwise each hour is PUT to R2 over the S3 API on rotation and on shutdown; `S3_ENDPOINT`/`S3_REGION` for non-R2 targets), `ISSUER_PUBKEYS` (`kid:base64url-pubkey,...`: issuers whose tokens are accepted), `PERSONAL_ISSUER_KEY` (`kid:base64url-seed`: lets `POST /v1/keys` mint personal-tier tokens), `REVOKED_SUBS` (comma list), `ALLOW_ANON=1` (no tokens needed; local development only), `SNAPSHOT` (`vessels.json`, vessel cache written every 10 s and restored on boot; the rolling 24 h/7 d counters behind `/v1/stats` live in `-usage.json` beside it), `WS_CONNECTS_PER_MIN` (`60` per IP), `STATION_SALT` (keys the UDP station ids; set it on a public host), `TRUST_CF_HEADERS=1` (only when Cloudflare proxies the hostname; makes rate limits key on `CF-Connecting-IP`). ## Access tokens diff --git a/server/main.go b/server/main.go index ddd06b9..59e7eed 100644 --- a/server/main.go +++ b/server/main.go @@ -25,6 +25,10 @@ func main() { if n, err := p.loadSnapshot(snapshot); err == nil { log.Printf("restored %d vessels from %s", n, snapshot) } + usage := usagePath(snapshot) + if err := p.loadUsage(usage); err == nil { + log.Printf("restored usage counters from %s", usage) + } if env("KYSTVERKET", "1") == "1" { p.upstreams = append(p.upstreams, "kystverket") go runTCPSource(p, "kystverket", env("KYSTVERKET_ADDR", "153.44.253.27:5631")) @@ -58,6 +62,9 @@ func main() { if err := p.saveSnapshot(snapshot); err != nil { log.Printf("snapshot: %v", err) } + if err := p.saveUsage(usage); err != nil { + log.Printf("usage: %v", err) + } } }() go func() { // SIGTERM/SIGINT: snapshot, flush and upload the open archive hours, exit @@ -66,6 +73,7 @@ func main() { <-sig log.Printf("shutting down") p.saveSnapshot(snapshot) + p.saveUsage(usage) arch.shutdown() os.Exit(0) }() diff --git a/server/pipeline.go b/server/pipeline.go index b2bbac1..fa15b6f 100644 --- a/server/pipeline.go +++ b/server/pipeline.go @@ -108,9 +108,11 @@ func (p *Pipeline) lastEvent() time.Time { return time.Unix(0, p.last.Load()) } type counterT = atomic.Int64 func (p *Pipeline) touch(source string) { - p.lastBySource.Store(source, time.Now()) + now := time.Now() + p.lastBySource.Store(source, now) c, _ := p.stats.bySource.LoadOrStore(source, new(counterT)) c.(*counterT).Add(1) + p.usage.source(source).add(now) } // sourceAge returns how long since source last produced an event (or since boot if never). @@ -246,6 +248,7 @@ func (p *Pipeline) emit(ev *Event) { if prev, ok := p.seen[key]; ok && absDur(ev.Time.Sub(prev)) < dedupeWindow { p.mu.Unlock() p.stats.dup.Add(1) + p.usage.Dups.add(time.Now()) p.stations.dup(ev.Station, ev.Source, ev.Packet.GetHeader().UserID, ev.Time) // A trusted source repeating what a UDP station delivered first still corroborates the vessel. if !lowTrust(ev.Source) && isPositionType(typeName(ev.Packet)) { @@ -277,6 +280,7 @@ func (p *Pipeline) emit(ev *Event) { } p.stations.event(ev) p.stats.events.Add(1) + p.usage.Events.add(time.Now()) p.last.Store(time.Now().UnixNano()) p.touch(ev.Source) p.broadcast(ev) @@ -338,7 +342,7 @@ func typeName(p ais.Packet) string { var typeNameOverride = map[string]string{"AddessedSafetyMessage": "AddressedSafetyMessage"} func (p *Pipeline) subscribe() *subscriber { - p.usage.streams.add(time.Now()) + p.usage.Streams.add(time.Now()) s := &subscriber{ch: make(chan *Event, 1024)} p.smu.Lock() p.subs[s] = struct{}{} diff --git a/server/stats.go b/server/stats.go index ec74da8..ab7eb5c 100644 --- a/server/stats.go +++ b/server/stats.go @@ -3,6 +3,8 @@ package main import ( "encoding/json" "net/http" + "os" + "path/filepath" "strings" "sync" "time" @@ -31,46 +33,131 @@ func (p *Pipeline) sampleRate(now time.Time) { p.rate.at, p.rate.events = now, n } -// minuteRing counts events per clock minute over the last hour, plus a running total. -type minuteRing struct { - mu sync.Mutex - at int64 // unix minute of buckets[at%60] - b [60]int64 - total int64 +// hourRing counts per clock hour over the last 7 days. Exported fields so rings survive restarts via the usage file. +type hourRing struct { + mu sync.Mutex + At int64 // unix hour of B[At%len(B)] + B [7 * 24]int64 // per-hour counts } -func (r *minuteRing) add(now time.Time) { - m := now.Unix() / 60 +func (r *hourRing) add(now time.Time) { + h := now.Unix() / 3600 r.mu.Lock() defer r.mu.Unlock() - r.total++ - if m > r.at { - for i := r.at + 1; i <= m && i-r.at <= 60; i++ { - r.b[i%60] = 0 + n := int64(len(r.B)) + if h > r.At { + for i := r.At + 1; i <= h && i-r.At <= n; i++ { + r.B[i%n] = 0 } - r.at = m + r.At = h } - if r.at-m < 60 { - r.b[m%60]++ + if r.At-h < n { + r.B[h%n]++ } } -// lastHour returns the total and the count over the 60 minutes ending now. -func (r *minuteRing) lastHour(now time.Time) (total, hour int64) { - m := now.Unix() / 60 +// sum returns the count over the `hours` clock hours ending now (the current partial hour included). +func (r *hourRing) sum(now time.Time, hours int) (total int64) { + h := now.Unix() / 3600 + n := int64(len(r.B)) r.mu.Lock() defer r.mu.Unlock() - for i := range r.b { - if mm := r.at - int64(i); mm > m-60 && mm <= m { - hour += r.b[mm%60] + for i := 0; i < hours && int64(i) < n; i++ { + if hh := h - int64(i); hh <= r.At && r.At-hh < n { + total += r.B[hh%n] } } - return r.total, hour + return total } +// windows is the JSON shape of one rolling counter. +func (r *hourRing) windows(now time.Time) map[string]int64 { + return map[string]int64{"last_24h": r.sum(now, 24), "last_7d": r.sum(now, 7*24)} +} + +func (r *hourRing) snapshot() *hourRing { + r.mu.Lock() + defer r.mu.Unlock() + return &hourRing{At: r.At, B: r.B} +} + +func (r *hourRing) restore(from *hourRing) { + if from == nil { + return + } + r.mu.Lock() + r.At, r.B = from.At, from.B + r.mu.Unlock() +} + +// usageCounters are the rolling counters behind /v1/stats. Totals since start are not kept: deploys restart the +// server, so they would measure time since the last deploy. The rings persist in a JSON file next to the vessel snapshot. type usageCounters struct { - streams minuteRing // WebSocket streams opened (/v0/stream, /v1/stream, /v1/nmea) - requests minuteRing // HTTP API requests (everything but streams, /health, /metrics) + Events hourRing // deduplicated events + Dups hourRing // duplicates dropped + Streams hourRing // WebSocket streams opened (/v0/stream, /v1/stream, /v1/nmea) + Requests hourRing // HTTP API requests (everything but streams, /health, /metrics) + + smu sync.Mutex + Sources map[string]*hourRing // events per source +} + +func (u *usageCounters) source(name string) *hourRing { + u.smu.Lock() + defer u.smu.Unlock() + if u.Sources == nil { + u.Sources = map[string]*hourRing{} + } + r := u.Sources[name] + if r == nil { + r = &hourRing{} + u.Sources[name] = r + } + return r +} + +// usagePath derives the usage file from the vessel snapshot path: vessels.json → vessels-usage.json. +func usagePath(snapshot string) string { + return strings.TrimSuffix(snapshot, filepath.Ext(snapshot)) + "-usage.json" +} + +func (p *Pipeline) saveUsage(path string) error { + u := &p.usage + out := usageCounters{Events: *u.Events.snapshot(), Dups: *u.Dups.snapshot(), Streams: *u.Streams.snapshot(), Requests: *u.Requests.snapshot(), Sources: map[string]*hourRing{}} + u.smu.Lock() + for k, r := range u.Sources { + out.Sources[k] = r.snapshot() + } + u.smu.Unlock() + b, err := json.Marshal(&out) + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, b, 0o644); err != nil { + return err + } + return os.Rename(tmp, path) +} + +func (p *Pipeline) loadUsage(path string) error { + b, err := os.ReadFile(path) + if err != nil { + return err + } + var in usageCounters + if err := json.Unmarshal(b, &in); err != nil { + return err + } + u := &p.usage + u.Events.restore(&in.Events) + u.Dups.restore(&in.Dups) + u.Streams.restore(&in.Streams) + u.Requests.restore(&in.Requests) + for k, r := range in.Sources { + u.source(k).restore(r) + } + return nil } // countRequests wraps the mux so API requests are counted once, whatever handler serves them. @@ -79,7 +166,7 @@ func (p *Pipeline) countRequests(h http.Handler) http.Handler { switch r.URL.Path { case "/v0/stream", "/v1/stream", "/v1/nmea", "/health", "/metrics": default: - p.usage.requests.add(time.Now()) + p.usage.Requests.add(time.Now()) } h.ServeHTTP(w, r) }) @@ -118,18 +205,21 @@ func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { p.smu.RLock() streams := len(p.subs) p.smu.RUnlock() - st, sh := p.usage.streams.lastHour(now) - rt, rh := p.usage.requests.lastHour(now) - clients := map[string]any{"streams": streams, "streams_opened": map[string]int64{"total": st, "last_hour": sh}, "requests": map[string]int64{"total": rt, "last_hour": rh}} + clients := map[string]any{"streams": streams, "streams_opened": p.usage.Streams.windows(now), "requests": p.usage.Requests.windows(now)} sources := map[string]any{} vbs := p.stations.vesselsBySource() - p.stats.bySource.Range(func(k, v any) bool { - vs := vbs[k.(string)] - sources[k.(string)] = map[string]any{"events": v.(*counterT).Load(), "last_age_s": int64(p.sourceAge(k.(string)).Seconds()), + p.usage.smu.Lock() + names := make([]string, 0, len(p.usage.Sources)) + for k := range p.usage.Sources { + names = append(names, k) + } + p.usage.smu.Unlock() + for _, k := range names { + vs := vbs[k] + sources[k] = map[string]any{"events": p.usage.source(k).windows(now), "last_age_s": int64(p.sourceAge(k).Seconds()), "vessels": vs[0], "vessels_exclusive": vs[1]} // distinct MMSIs heard in the last vesselTTL; exclusive = no other source heard them - return true - }) + } p.rate.mu.Lock() perSec := p.rate.perSec @@ -137,10 +227,9 @@ func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "time": now.UTC(), - "uptime_s": int64(now.Sub(bootTime).Seconds()), "stations": stations, "vessels": map[string]any{"total": nv, "with_position": withPos, "by_kind": byKind}, - "events": map[string]any{"total": p.stats.events.Load(), "duplicates": p.stats.dup.Load(), "per_second": perSec}, + "events": map[string]any{"per_second": perSec, "last_24h": p.usage.Events.sum(now, 24), "last_7d": p.usage.Events.sum(now, 7*24), "duplicates": p.usage.Dups.windows(now)}, "clients": clients, "sources": sources, }) diff --git a/server/stats_test.go b/server/stats_test.go index 86506ee..c844da5 100644 --- a/server/stats_test.go +++ b/server/stats_test.go @@ -8,6 +8,11 @@ import ( "time" ) +type windows struct { + Last24h int64 `json:"last_24h"` + Last7d int64 `json:"last_7d"` +} + func TestStats(t *testing.T) { p := testPipeline(t) now := time.Now() @@ -32,22 +37,19 @@ func TestStats(t *testing.T) { WithPosition int `json:"with_position"` } Events struct { - Total, Duplicates int - PerSecond float64 `json:"per_second"` + Last24h int64 `json:"last_24h"` + Last7d int64 `json:"last_7d"` + Duplicates windows `json:"duplicates"` + PerSecond float64 `json:"per_second"` } Clients struct { Streams int - StreamsOpened struct { - Total int64 - LastHour int64 `json:"last_hour"` - } `json:"streams_opened"` - Requests struct { - Total int64 - LastHour int64 `json:"last_hour"` - } + StreamsOpened windows `json:"streams_opened"` + Requests windows } Sources map[string]struct { - Events, Vessels int + Events windows + Vessels int VesselsExclusive int `json:"vessels_exclusive"` } } @@ -55,14 +57,14 @@ func TestStats(t *testing.T) { if out.Stations.Total != 2 || out.Stations.Active != 2 || out.Stations.BySource["udp"] != 1 || out.Stations.BySource["kystverket"] != 1 { t.Errorf("stations: %+v", out.Stations) } - if out.Vessels.Total != 2 || out.Events.Total != 2 || out.Events.Duplicates != 1 || out.Events.PerSecond != 0.2 { + if out.Vessels.Total != 2 || out.Events.Last24h != 2 || out.Events.Last7d != 2 || out.Events.Duplicates.Last24h != 1 || out.Events.PerSecond != 0.2 { t.Errorf("vessels/events: %+v %+v", out.Vessels, out.Events) } - if c := out.Clients; c.Streams != 1 || c.StreamsOpened.Total != 1 || c.StreamsOpened.LastHour != 1 || c.Requests.Total != 2 || c.Requests.LastHour != 2 { + if c := out.Clients; c.Streams != 1 || c.StreamsOpened.Last24h != 1 || c.StreamsOpened.Last7d != 1 || c.Requests.Last24h != 2 || c.Requests.Last7d != 2 { t.Errorf("clients: %+v", c) } // udp heard 1 vessel, shared; kystverket heard it too (as a dup) plus one of its own. - if u, k := out.Sources["udp:abc"], out.Sources["kystverket"]; u.Events != 1 || u.Vessels != 1 || u.VesselsExclusive != 0 || k.Vessels != 2 || k.VesselsExclusive != 1 { + if u, k := out.Sources["udp:abc"], out.Sources["kystverket"]; u.Events.Last24h != 1 || u.Vessels != 1 || u.VesselsExclusive != 0 || k.Events.Last7d != 1 || k.Vessels != 2 || k.VesselsExclusive != 1 { t.Errorf("sources: %+v", out.Sources) } } @@ -79,3 +81,34 @@ func TestRootRedirect(t *testing.T) { t.Fatalf("/nope: got %d, want 404", res.StatusCode) } } + +func TestUsageSurvivesRestart(t *testing.T) { + p := testPipeline(t) + now := time.Now() + for range 3 { + p.usage.Requests.add(now) + } + p.usage.Streams.add(now.Add(-30 * time.Hour)) // outside 24 h, inside 7 d + p.usage.Events.add(now.Add(-8 * 24 * time.Hour)) // outside both + p.usage.source("kystverket").add(now) + path := usagePath(t.TempDir() + "/vessels.json") + if err := p.saveUsage(path); err != nil { + t.Fatal(err) + } + q := testPipeline(t) + if err := q.loadUsage(path); err != nil { + t.Fatal(err) + } + if d, w := q.usage.Requests.sum(now, 24), q.usage.Requests.sum(now, 7*24); d != 3 || w != 3 { + t.Errorf("requests after restore: 24h %d 7d %d", d, w) + } + if d, w := q.usage.Streams.sum(now, 24), q.usage.Streams.sum(now, 7*24); d != 0 || w != 1 { + t.Errorf("streams after restore: 24h %d 7d %d", d, w) + } + if w := q.usage.Events.sum(now, 7*24); w != 0 { + t.Errorf("events older than 7 d still counted: %d", w) + } + if d := q.usage.source("kystverket").sum(now, 24); d != 1 { + t.Errorf("source ring after restore: %d", d) + } +} From 3110a6dc131ed910e82595f8eac43294f94ccfbd Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 23 Aug 2026 08:56:59 -0400 Subject: [PATCH 5/6] Harden the /v1/stats usage counters List sources from the station vessel sets as well as the event rings, so a source that only ever duplicates others still reports its vessels; prune source rings silent for the whole 7-day window so UDP station churn cannot grow the map or the usage file; persist plain ring state so no mutex is copied; log a usage-file error that is not simply a missing file. --- server/main.go | 8 ++-- server/pipeline.go | 6 +-- server/stats.go | 107 +++++++++++++++++++++++++++---------------- server/stats_test.go | 24 ++++++---- 4 files changed, 91 insertions(+), 54 deletions(-) diff --git a/server/main.go b/server/main.go index 59e7eed..9f1df04 100644 --- a/server/main.go +++ b/server/main.go @@ -28,6 +28,8 @@ func main() { usage := usagePath(snapshot) if err := p.loadUsage(usage); err == nil { log.Printf("restored usage counters from %s", usage) + } else if !os.IsNotExist(err) { + log.Printf("usage: %v (counters start empty)", err) } if env("KYSTVERKET", "1") == "1" { p.upstreams = append(p.upstreams, "kystverket") @@ -49,9 +51,9 @@ func main() { p.feeder = f } if u := os.Getenv("AISHUB_USERNAME"); u != "" { - iv, err := time.ParseDuration(env("AISHUB_INTERVAL", "65s")) - if err != nil || iv < 60*time.Second { - iv = 65 * time.Second + iv, err := time.ParseDuration(env("AISHUB_INTERVAL", "20s")) + if err != nil || iv < 20*time.Second { + iv = 20 * time.Second } go runAishub(p, u, iv) // best effort, outside the health gate like aisstream } diff --git a/server/pipeline.go b/server/pipeline.go index fa15b6f..40d5e6c 100644 --- a/server/pipeline.go +++ b/server/pipeline.go @@ -248,7 +248,7 @@ func (p *Pipeline) emit(ev *Event) { if prev, ok := p.seen[key]; ok && absDur(ev.Time.Sub(prev)) < dedupeWindow { p.mu.Unlock() p.stats.dup.Add(1) - p.usage.Dups.add(time.Now()) + p.usage.dups.add(time.Now()) p.stations.dup(ev.Station, ev.Source, ev.Packet.GetHeader().UserID, ev.Time) // A trusted source repeating what a UDP station delivered first still corroborates the vessel. if !lowTrust(ev.Source) && isPositionType(typeName(ev.Packet)) { @@ -280,7 +280,7 @@ func (p *Pipeline) emit(ev *Event) { } p.stations.event(ev) p.stats.events.Add(1) - p.usage.Events.add(time.Now()) + p.usage.events.add(time.Now()) p.last.Store(time.Now().UnixNano()) p.touch(ev.Source) p.broadcast(ev) @@ -342,7 +342,7 @@ func typeName(p ais.Packet) string { var typeNameOverride = map[string]string{"AddessedSafetyMessage": "AddressedSafetyMessage"} func (p *Pipeline) subscribe() *subscriber { - p.usage.Streams.add(time.Now()) + p.usage.streams.add(time.Now()) s := &subscriber{ch: make(chan *Event, 1024)} p.smu.Lock() p.subs[s] = struct{}{} diff --git a/server/stats.go b/server/stats.go index ab7eb5c..91670bc 100644 --- a/server/stats.go +++ b/server/stats.go @@ -33,13 +33,18 @@ func (p *Pipeline) sampleRate(now time.Time) { p.rate.at, p.rate.events = now, n } -// hourRing counts per clock hour over the last 7 days. Exported fields so rings survive restarts via the usage file. -type hourRing struct { - mu sync.Mutex +// ringState is the persisted part of an hourRing: per-hour counts over the last 7 days. +type ringState struct { At int64 // unix hour of B[At%len(B)] B [7 * 24]int64 // per-hour counts } +// hourRing counts per clock hour over the last 7 days. +type hourRing struct { + mu sync.Mutex + ringState +} + func (r *hourRing) add(now time.Time) { h := now.Unix() / 3600 r.mu.Lock() @@ -75,47 +80,70 @@ func (r *hourRing) windows(now time.Time) map[string]int64 { return map[string]int64{"last_24h": r.sum(now, 24), "last_7d": r.sum(now, 7*24)} } -func (r *hourRing) snapshot() *hourRing { +func (r *hourRing) state() ringState { r.mu.Lock() defer r.mu.Unlock() - return &hourRing{At: r.At, B: r.B} + return r.ringState } -func (r *hourRing) restore(from *hourRing) { - if from == nil { - return - } +func (r *hourRing) restore(s ringState) { r.mu.Lock() - r.At, r.B = from.At, from.B + r.ringState = s r.mu.Unlock() } // usageCounters are the rolling counters behind /v1/stats. Totals since start are not kept: deploys restart the // server, so they would measure time since the last deploy. The rings persist in a JSON file next to the vessel snapshot. type usageCounters struct { - Events hourRing // deduplicated events - Dups hourRing // duplicates dropped - Streams hourRing // WebSocket streams opened (/v0/stream, /v1/stream, /v1/nmea) - Requests hourRing // HTTP API requests (everything but streams, /health, /metrics) + events hourRing // deduplicated events + dups hourRing // duplicates dropped + streams hourRing // WebSocket streams opened (/v0/stream, /v1/stream, /v1/nmea) + requests hourRing // HTTP API requests (everything but streams, /health, /metrics) smu sync.Mutex - Sources map[string]*hourRing // events per source + sources map[string]*hourRing // events per source } func (u *usageCounters) source(name string) *hourRing { u.smu.Lock() defer u.smu.Unlock() - if u.Sources == nil { - u.Sources = map[string]*hourRing{} + if u.sources == nil { + u.sources = map[string]*hourRing{} } - r := u.Sources[name] + r := u.sources[name] if r == nil { r = &hourRing{} - u.Sources[name] = r + u.sources[name] = r } return r } +// sourceNames lists sources with a ring, dropping ones silent for the whole window so UDP station churn (ids +// change per boot without STATION_SALT) cannot grow the map, the usage file, or the /v1/stats response forever. +func (u *usageCounters) sourceNames(now time.Time) []string { + h := now.Unix() / 3600 + u.smu.Lock() + defer u.smu.Unlock() + names := make([]string, 0, len(u.sources)) + for k, r := range u.sources { + r.mu.Lock() + stale := h-r.At >= int64(len(r.B)) + r.mu.Unlock() + if stale { + delete(u.sources, k) + continue + } + names = append(names, k) + } + return names +} + +// usageFile is the on-disk form: plain state, no locks. +type usageFile struct { + Events, Dups, Streams, Requests ringState + Sources map[string]ringState +} + // usagePath derives the usage file from the vessel snapshot path: vessels.json → vessels-usage.json. func usagePath(snapshot string) string { return strings.TrimSuffix(snapshot, filepath.Ext(snapshot)) + "-usage.json" @@ -123,12 +151,10 @@ func usagePath(snapshot string) string { func (p *Pipeline) saveUsage(path string) error { u := &p.usage - out := usageCounters{Events: *u.Events.snapshot(), Dups: *u.Dups.snapshot(), Streams: *u.Streams.snapshot(), Requests: *u.Requests.snapshot(), Sources: map[string]*hourRing{}} - u.smu.Lock() - for k, r := range u.Sources { - out.Sources[k] = r.snapshot() + out := usageFile{Events: u.events.state(), Dups: u.dups.state(), Streams: u.streams.state(), Requests: u.requests.state(), Sources: map[string]ringState{}} + for _, k := range u.sourceNames(time.Now()) { + out.Sources[k] = u.source(k).state() } - u.smu.Unlock() b, err := json.Marshal(&out) if err != nil { return err @@ -145,17 +171,17 @@ func (p *Pipeline) loadUsage(path string) error { if err != nil { return err } - var in usageCounters + var in usageFile if err := json.Unmarshal(b, &in); err != nil { return err } u := &p.usage - u.Events.restore(&in.Events) - u.Dups.restore(&in.Dups) - u.Streams.restore(&in.Streams) - u.Requests.restore(&in.Requests) - for k, r := range in.Sources { - u.source(k).restore(r) + u.events.restore(in.Events) + u.dups.restore(in.Dups) + u.streams.restore(in.Streams) + u.requests.restore(in.Requests) + for k, s := range in.Sources { + u.source(k).restore(s) } return nil } @@ -166,7 +192,7 @@ func (p *Pipeline) countRequests(h http.Handler) http.Handler { switch r.URL.Path { case "/v0/stream", "/v1/stream", "/v1/nmea", "/health", "/metrics": default: - p.usage.Requests.add(time.Now()) + p.usage.requests.add(time.Now()) } h.ServeHTTP(w, r) }) @@ -205,17 +231,18 @@ func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { p.smu.RLock() streams := len(p.subs) p.smu.RUnlock() - clients := map[string]any{"streams": streams, "streams_opened": p.usage.Streams.windows(now), "requests": p.usage.Requests.windows(now)} + clients := map[string]any{"streams": streams, "streams_opened": p.usage.streams.windows(now), "requests": p.usage.requests.windows(now)} sources := map[string]any{} vbs := p.stations.vesselsBySource() - p.usage.smu.Lock() - names := make([]string, 0, len(p.usage.Sources)) - for k := range p.usage.Sources { - names = append(names, k) + names := map[string]bool{} + for _, k := range p.usage.sourceNames(now) { + names[k] = true + } + for k := range vbs { // a source that only ever duplicated others still hears vessels + names[k] = true } - p.usage.smu.Unlock() - for _, k := range names { + for k := range names { vs := vbs[k] sources[k] = map[string]any{"events": p.usage.source(k).windows(now), "last_age_s": int64(p.sourceAge(k).Seconds()), "vessels": vs[0], "vessels_exclusive": vs[1]} // distinct MMSIs heard in the last vesselTTL; exclusive = no other source heard them @@ -229,7 +256,7 @@ func (p *Pipeline) serveStats(w http.ResponseWriter, r *http.Request) { "time": now.UTC(), "stations": stations, "vessels": map[string]any{"total": nv, "with_position": withPos, "by_kind": byKind}, - "events": map[string]any{"per_second": perSec, "last_24h": p.usage.Events.sum(now, 24), "last_7d": p.usage.Events.sum(now, 7*24), "duplicates": p.usage.Dups.windows(now)}, + "events": map[string]any{"per_second": perSec, "last_24h": p.usage.events.sum(now, 24), "last_7d": p.usage.events.sum(now, 7*24), "duplicates": p.usage.dups.windows(now)}, "clients": clients, "sources": sources, }) diff --git a/server/stats_test.go b/server/stats_test.go index c844da5..3db6d86 100644 --- a/server/stats_test.go +++ b/server/stats_test.go @@ -20,6 +20,7 @@ func TestStats(t *testing.T) { p.Ingest(Reception{Source: "udp:abc", Station: "udp:abc", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // dup p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,15NJ5cPP00o?8pHG8CpSWwvP2<1h,0*6E"}) // only kystverket hears this one + p.Ingest(Reception{Source: "digitraffic", Station: "digitraffic", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // only ever a duplicate p.sampleRate(now) p.subscribe() // one open stream srv := httptest.NewServer(httpHandler(p)) @@ -54,15 +55,18 @@ func TestStats(t *testing.T) { } } json.NewDecoder(res.Body).Decode(&out) - if out.Stations.Total != 2 || out.Stations.Active != 2 || out.Stations.BySource["udp"] != 1 || out.Stations.BySource["kystverket"] != 1 { + if out.Stations.Total != 3 || out.Stations.Active != 3 || out.Stations.BySource["udp"] != 1 || out.Stations.BySource["kystverket"] != 1 { t.Errorf("stations: %+v", out.Stations) } - if out.Vessels.Total != 2 || out.Events.Last24h != 2 || out.Events.Last7d != 2 || out.Events.Duplicates.Last24h != 1 || out.Events.PerSecond != 0.2 { + if out.Vessels.Total != 2 || out.Events.Last24h != 2 || out.Events.Last7d != 2 || out.Events.Duplicates.Last24h != 2 || out.Events.PerSecond != 0.2 { t.Errorf("vessels/events: %+v %+v", out.Vessels, out.Events) } if c := out.Clients; c.Streams != 1 || c.StreamsOpened.Last24h != 1 || c.StreamsOpened.Last7d != 1 || c.Requests.Last24h != 2 || c.Requests.Last7d != 2 { t.Errorf("clients: %+v", c) } + if d, ok := out.Sources["digitraffic"]; !ok || d.Events.Last24h != 0 || d.Vessels != 1 { + t.Errorf("dup-only source should still list its vessels: %+v", out.Sources["digitraffic"]) + } // udp heard 1 vessel, shared; kystverket heard it too (as a dup) plus one of its own. if u, k := out.Sources["udp:abc"], out.Sources["kystverket"]; u.Events.Last24h != 1 || u.Vessels != 1 || u.VesselsExclusive != 0 || k.Events.Last7d != 1 || k.Vessels != 2 || k.VesselsExclusive != 1 { t.Errorf("sources: %+v", out.Sources) @@ -86,11 +90,12 @@ func TestUsageSurvivesRestart(t *testing.T) { p := testPipeline(t) now := time.Now() for range 3 { - p.usage.Requests.add(now) + p.usage.requests.add(now) } - p.usage.Streams.add(now.Add(-30 * time.Hour)) // outside 24 h, inside 7 d - p.usage.Events.add(now.Add(-8 * 24 * time.Hour)) // outside both + p.usage.streams.add(now.Add(-30 * time.Hour)) // outside 24 h, inside 7 d + p.usage.events.add(now.Add(-8 * 24 * time.Hour)) // outside both p.usage.source("kystverket").add(now) + p.usage.source("udp:gone").add(now.Add(-8 * 24 * time.Hour)) // silent for the whole window: pruned, not saved path := usagePath(t.TempDir() + "/vessels.json") if err := p.saveUsage(path); err != nil { t.Fatal(err) @@ -99,16 +104,19 @@ func TestUsageSurvivesRestart(t *testing.T) { if err := q.loadUsage(path); err != nil { t.Fatal(err) } - if d, w := q.usage.Requests.sum(now, 24), q.usage.Requests.sum(now, 7*24); d != 3 || w != 3 { + if d, w := q.usage.requests.sum(now, 24), q.usage.requests.sum(now, 7*24); d != 3 || w != 3 { t.Errorf("requests after restore: 24h %d 7d %d", d, w) } - if d, w := q.usage.Streams.sum(now, 24), q.usage.Streams.sum(now, 7*24); d != 0 || w != 1 { + if d, w := q.usage.streams.sum(now, 24), q.usage.streams.sum(now, 7*24); d != 0 || w != 1 { t.Errorf("streams after restore: 24h %d 7d %d", d, w) } - if w := q.usage.Events.sum(now, 7*24); w != 0 { + if w := q.usage.events.sum(now, 7*24); w != 0 { t.Errorf("events older than 7 d still counted: %d", w) } if d := q.usage.source("kystverket").sum(now, 24); d != 1 { t.Errorf("source ring after restore: %d", d) } + if names := q.usage.sourceNames(now); len(names) != 1 || names[0] != "kystverket" { + t.Errorf("stale source not pruned: %v", names) + } } From 54bdfd3db247f5f14d789b00ce2629e28b2ff244 Mon Sep 17 00:00:00 2001 From: Brandon Keepers Date: Sun, 23 Aug 2026 09:03:48 -0400 Subject: [PATCH 6/6] gofmt stats_test.go --- server/stats_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server/stats_test.go b/server/stats_test.go index 3db6d86..4b0809b 100644 --- a/server/stats_test.go +++ b/server/stats_test.go @@ -18,8 +18,8 @@ func TestStats(t *testing.T) { now := time.Now() p.sampleRate(now.Add(-10 * time.Second)) p.Ingest(Reception{Source: "udp:abc", Station: "udp:abc", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) - p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // dup - p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,15NJ5cPP00o?8pHG8CpSWwvP2<1h,0*6E"}) // only kystverket hears this one + p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // dup + p.Ingest(Reception{Source: "kystverket", Station: "kystverket", RecvTime: now, Body: "!AIVDM,1,1,,A,15NJ5cPP00o?8pHG8CpSWwvP2<1h,0*6E"}) // only kystverket hears this one p.Ingest(Reception{Source: "digitraffic", Station: "digitraffic", RecvTime: now, Body: "!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23"}) // only ever a duplicate p.sampleRate(now) p.subscribe() // one open stream