Preserve source identity and history references through layer state

src/sources/live/contract.test.mjs

@@ -247,3 +247,17 @@ test('AIS reports limited received coverage and keeps connection state separate
     now,
   );
 });
+
+test('a malformed vessel row cannot prevent admission of valid positions', async () => {
+  const source = createAisStreamSource({
+    fetchImpl: async () =>
+      response({
+        status: 'live',
+        rows: [null, { mmsi: '123456789', lat: 30, lon: -97 }],
+      }),
+  });
+  const snapshot = await source.getSnapshot();
+  assert.equal(snapshot.records.length, 1);
+  assert.equal(snapshot.rejectedCount, 1);
+  assert.equal(snapshot.complete, false);
+});

src/sources/live/standalone.js

@@ -54,6 +54,7 @@ export function createOpenSkySource({
   now = () => Date.now(),
 } = {}) {
   return {
+    label: 'OpenSky Network',
     async getSnapshot(query = {}, { signal } = {}) {
       const params = new URLSearchParams();
       if (Number.isFinite(query.latitude) && Number.isFinite(query.longitude)) {
@@ -111,6 +112,7 @@ export function createAdsbLolSource({
   now = () => Date.now(),
 } = {}) {
   return {
+    label: 'adsb.lol',
     async getSnapshot(_query = {}, { signal } = {}) {
       const { response, payload } = await readResponse(
         fetchImpl,
@@ -158,6 +160,7 @@ export function createAisStreamSource({
   origin = () => globalThis.location?.origin || 'http://localhost',
 } = {}) {
   return {
+    label: 'AISStream',
     async getSnapshot({ maxRows = 12000 } = {}, { signal } = {}) {
       const url = new URL(apiUrl, origin());
       url.searchParams.set('maxRows', String(maxRows));

src/sources/live/vessels.js

@@ -40,7 +40,8 @@ export function vesselSnapshot(
   const records = [],
     ids = new Set();
   for (const row of rows) {
-    const record = normalizeVesselObservation(row, referenceFor(row));
+    const record = normalizeVesselObservation(row);
+    if (record) record.reference = referenceFor(row);
     if (record && !ids.has(record.id)) {
       records.push(record);
       ids.add(record.id);
@@ -57,7 +58,7 @@ export function vesselSnapshot(
     records,
     source,
     coverage,
-    complete: false,
+    complete: records.length === rows.length && !payload?.refreshing,
     rejectedCount: rows.length - records.length,
     observedAtMs,
     freshness: payload?.refreshing