Prevent deleted documents from being resurrected in the runtime cache

.github/workflows/build-image.yml

@@ -0,0 +1,126 @@
+# Build and publish this fork to ghcr.io/bjorkert/nightscout, matching the
+# pattern used by the other bjorkert images (nsgram, cart, lftelemetryapi).
+#
+# Two things differ from those repos on purpose:
+#
+#  1. NO version bump commit. Those repos own their version file; this one is a
+#     fork that keeps merging upstream nightscout/cgm-remote-monitor, and
+#     rewriting package.json in CI would create a conflict on every single
+#     upstream merge. The image is tagged from the version already in
+#     package.json plus the commit SHA instead.
+#
+#  2. Native per-arch runners instead of QEMU. This repo is public, so
+#     ubuntu-24.04-arm is free. Nightscout's `npm ci` builds native modules;
+#     under QEMU emulation the arm64 leg takes the better part of an hour,
+#     versus a few minutes natively. Each arch is built and pushed by digest,
+#     then a merge job stitches them into one multi-arch manifest.
+name: Build and Push Nightscout Image
+
+on:
+  push:
+    # Deploy branch, not master: master tracks upstream nightscout, and what we
+    # actually run is this fork's work. NOTE this is a topic branch — if it ever
+    # gets merged and deleted, pushes stop triggering builds silently and
+    # ghcr.io/bjorkert/nightscout:latest quietly goes stale. Point this at
+    # whatever branch is being deployed at the time.
+    branches: [fix/cache-stale-delete-race]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+  packages: write
+
+env:
+  IMAGE_NAME: ghcr.io/bjorkert/nightscout
+
+jobs:
+  build:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - platform: linux/amd64
+            runner: ubuntu-24.04
+          - platform: linux/arm64
+            runner: ubuntu-24.04-arm
+    runs-on: ${{ matrix.runner }}
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v4
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v4
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Build and push by digest
+        id: build
+        uses: docker/build-push-action@v7
+        with:
+          context: .
+          file: ./Dockerfile
+          platforms: ${{ matrix.platform }}
+          # push-by-digest publishes an untagged image; the merge job below is
+          # what creates the human-readable tags. Without this, the two arch
+          # jobs would race and each overwrite the other's :latest.
+          outputs: type=image,name=${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
+          cache-from: type=gha,scope=${{ matrix.platform }}
+          cache-to: type=gha,mode=max,scope=${{ matrix.platform }}
+
+      - name: Export digest
+        # The stripping has to happen in bash — GitHub Actions expressions do
+        # not support ${var#prefix} parameter expansion.
+        run: |
+          mkdir -p /tmp/digests
+          digest="${{ steps.build.outputs.digest }}"
+          touch "/tmp/digests/${digest#sha256:}"
+
+      - name: Upload digest
+        uses: actions/upload-artifact@v4
+        with:
+          name: digest-${{ strategy.job-index }}
+          path: /tmp/digests/*
+          retention-days: 1
+
+  merge:
+    needs: [build]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Download digests
+        uses: actions/download-artifact@v4
+        with:
+          path: /tmp/digests
+          pattern: digest-*
+          merge-multiple: true
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v4
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v4
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Read version from package.json
+        id: ver
+        run: echo "version=$(node -p "require('./package.json').version")" >> "$GITHUB_OUTPUT"
+
+      - name: Create multi-arch manifest
+        working-directory: /tmp/digests
+        run: |
+          docker buildx imagetools create \
+            -t ${IMAGE_NAME}:latest \
+            -t ${IMAGE_NAME}:${{ steps.ver.outputs.version }} \
+            -t ${IMAGE_NAME}:sha-${GITHUB_SHA::7} \
+            $(printf "${IMAGE_NAME}@sha256:%s " *)
+
+      - name: Show result
+        run: docker buildx imagetools inspect ${IMAGE_NAME}:latest

lib/data/dataloader.js

@@ -150,11 +150,13 @@ function init(env, ctx) {
 
 }
 
-function loadEntries(ddata, ctx, callback) {
+function loadEntries(ddata, ctx, callback, attempt) {
 
+    attempt = attempt || 0;
     const withFrame = ddata.page && ddata.page.frame;
     const longLoad = Math.round(constants.TWO_DAYS);
     const loadTime = ctx.cache.isEmpty('entries') || withFrame ? longLoad : constants.FIFTEEN_MINUTES;
+    const cacheGeneration = ctx.cache.getRemovalGeneration('entries');
 
     var dateRange = {
         $gte: ddata.lastUpdated - loadTime
@@ -180,8 +182,17 @@ function loadEntries(ddata, ctx, callback) {
 
         if (!err && results) {
 
+            // a removal invalidated the cache while the query was in flight;
+            // the results may contain deleted documents, so reload instead of
+            // merging them into the cache
+            if (ctx.cache.getRemovalGeneration('entries') !== cacheGeneration && attempt < 2) {
+                return loadEntries(ddata, ctx, callback, attempt + 1);
+            }
+
             const r = ctx.ddata.processRawDataForRuntime(results);
-            const currentData = ctx.cache.insertData('entries', r).reverse();
+            const currentData = ctx.cache.getRemovalGeneration('entries') !== cacheGeneration
+                ? ctx.cache.getData('entries').reverse()
+                : ctx.cache.insertData('entries', r).reverse();
 
             const mbgs = [];
             const sgvs = [];
@@ -277,15 +288,17 @@ function loadActivity(ddata, ctx, callback) {
     });
 }
 
-function loadTreatments(ddata, ctx, callback) {
+function loadTreatments(ddata, ctx, callback, attempt) {
 
+    attempt = attempt || 0;
     const withFrame = ddata.page && ddata.page.frame;
     const longLoad = Math.round(constants.ONE_DAY * 2.5); //ONE_DAY * 2.5;
 
     // Load 2.5 days to cover last 48 hours including overlapping temp boluses or temp targets for first load
     // Subsequently load at least 15 minutes of data
 
     const loadTime = ctx.cache.isEmpty('treatments') || withFrame ? longLoad : constants.FIFTEEN_MINUTES;
+    const cacheGeneration = ctx.cache.getRemovalGeneration('treatments');
 
     var dateRange = {
         $gte: new Date(ddata.lastUpdated - loadTime).toISOString()
@@ -305,9 +318,18 @@ function loadTreatments(ddata, ctx, callback) {
     ctx.treatments.list(tq, function(err, results) {
         if (!err && results) {
 
+            // a removal invalidated the cache while the query was in flight;
+            // the results may contain deleted documents, so reload instead of
+            // merging them into the cache
+            if (ctx.cache.getRemovalGeneration('treatments') !== cacheGeneration && attempt < 2) {
+                return loadTreatments(ddata, ctx, callback, attempt + 1);
+            }
+
             // update cache and apply to runtime data
             const r = ctx.ddata.processRawDataForRuntime(results);
-            const currentData = ctx.cache.insertData('treatments', r);
+            const currentData = ctx.cache.getRemovalGeneration('treatments') !== cacheGeneration
+                ? ctx.cache.getData('treatments')
+                : ctx.cache.insertData('treatments', r);
             ddata.treatments = ctx.ddata.idMergePreferNew(ddata.treatments, currentData);
         }
 
@@ -425,11 +447,13 @@ function loadFood(ddata, ctx, callback) {
     });
 }
 
-function loadDeviceStatus(ddata, env, ctx, callback) {
+function loadDeviceStatus(ddata, env, ctx, callback, attempt) {
 
+    attempt = attempt || 0;
     const withFrame = ddata.page && ddata.page.frame;
     const longLoad = env.extendedSettings.devicestatus && env.extendedSettings.devicestatus.days && env.extendedSettings.devicestatus.days == 2 ? constants.TWO_DAYS : constants.ONE_DAY;
     const loadTime = ctx.cache.isEmpty('devicestatus') || withFrame ? longLoad : constants.FIFTEEN_MINUTES;
+    const cacheGeneration = ctx.cache.getRemovalGeneration('devicestatus');
 
     var dateRange = {
         $gte: new Date( ddata.lastUpdated -  loadTime ).toISOString()
@@ -451,9 +475,18 @@ function loadDeviceStatus(ddata, env, ctx, callback) {
     ctx.devicestatus.list(opts, function(err, results) {
         if (!err && results) {
 
+            // a removal invalidated the cache while the query was in flight;
+            // the results may contain deleted documents, so reload instead of
+            // merging them into the cache
+            if (ctx.cache.getRemovalGeneration('devicestatus') !== cacheGeneration && attempt < 2) {
+                return loadDeviceStatus(ddata, env, ctx, callback, attempt + 1);
+            }
+
             // update cache and apply to runtime data
             const r = ctx.ddata.processRawDataForRuntime(results);
-            const currentData = ctx.cache.insertData('devicestatus', r);
+            const currentData = ctx.cache.getRemovalGeneration('devicestatus') !== cacheGeneration
+                ? ctx.cache.getData('devicestatus')
+                : ctx.cache.insertData('devicestatus', r);
             const res2 = currentData.map(function eachStatus(result) {
                 if ('uploaderBattery' in result) {
                     result.uploader = {

lib/server/cache.js

@@ -29,6 +29,19 @@ function cache (env, ctx) {
     , entries: constants.TWO_DAYS
   };
 
+  /* Each removal bumps the generation counter for the affected datatype.
+   * The dataloader reads the counter before querying Mongo, so it can tell
+   * when a delete landed while a query was in flight and the results may
+   * still contain the deleted documents. Merging such results into the
+   * cache would resurrect them until the retention period expires, since
+   * incremental loads never revisit that time window.
+   */
+  const removalGenerations = {
+    treatments: 0
+    , devicestatus: 0
+    , entries: 0
+  };
+
   function getObjectAge(object) {
     let age = object.mills || object.date;
     if (isNaN(age) && object.created_at) age = Date.parse(object.created_at).valueOf();
@@ -68,6 +81,10 @@ function cache (env, ctx) {
     return data.getData(datatype);
   }
 
+  data.getRemovalGeneration = (datatype) => {
+    return removalGenerations[datatype];
+  }
+
   function dataChanged (operation) {
     if (!data[operation.type]) return;
 
@@ -77,8 +94,12 @@ function cache (env, ctx) {
         data.treatments = [];
         data.devicestatus = [];
         data.entries = [];
+        removalGenerations.treatments += 1;
+        removalGenerations.devicestatus += 1;
+        removalGenerations.entries += 1;
       } else {
         removeFromArray(data[operation.type], operation.changes);
+        removalGenerations[operation.type] += 1;
       }
     }
 

tests/dataloader.test.js

@@ -16,7 +16,8 @@ describe('dataloader', function () {
       },
       cache: {
         isEmpty: function () { return true; },
-        insertData: function (key, results) { return results; }
+        insertData: function (key, results) { return results; },
+        getRemovalGeneration: function () { return 0; }
       },
       ddata: ddata,
       entries: {
@@ -63,4 +64,81 @@ describe('dataloader', function () {
       done();
     });
   });
+
+  it('does not resurrect treatments deleted while a load is in flight', function (done) {
+    const ddata = createDData();
+    ddata.processTreatments = function () {};
+
+    const bus = new (require('events').EventEmitter)();
+    const ghost = {
+      _id: 'aaaaaaaaaaaaaaaaaaaaaaaa',
+      eventType: 'Exercise',
+      created_at: new Date(Date.now() - 60000).toISOString(),
+      mills: Date.now() - 60000,
+      duration: 43200
+    };
+
+    let treatmentLoads = 0;
+    const ctx = {
+      settings: {},
+      bus: bus,
+      language: {
+        translate: function (value) { return value; }
+      },
+      ddata: ddata,
+      entries: {
+        list: function (query, callback) { callback(null, []); }
+      },
+      treatments: {
+        list: function (query, callback) {
+          // secondary loaders filter on eventType; only the main load matters here
+          if (query.find && query.find.eventType) return callback(null, []);
+          treatmentLoads += 1;
+          if (treatmentLoads === 1) {
+            // the query result still contains the document, but the delete
+            // commits and flushes the cache before the merge happens
+            bus.emit('data-update', { type: 'treatments', op: 'remove', count: 1 });
+            return callback(null, [ghost]);
+          }
+          callback(null, []);
+        }
+      },
+      profile: {
+        last: function (callback) { callback(null, []); }
+      },
+      food: {
+        list: function (callback) { callback(null, []); }
+      },
+      devicestatus: {
+        list: function (query, callback) { callback(null, []); }
+      },
+      activity: {
+        list: function (query, callback) { callback(null, []); }
+      },
+      store: {
+        db: {
+          stats: function () {
+            return Promise.resolve({ dataSize: 123, indexSize: 456 });
+          }
+        }
+      }
+    };
+    const env = {
+      settings: {
+        isEnabled: function () { return false; },
+        units: 'mg/dl'
+      },
+      extendedSettings: {}
+    };
+    ctx.cache = require('../lib/server/cache')(env, ctx);
+    const loader = dataloaderInit(env, ctx);
+
+    loader.update(ddata, function (err) {
+      should.not.exist(err);
+      treatmentLoads.should.equal(2);
+      ddata.treatments.filter(function (t) { return t._id === ghost._id; }).should.have.length(0);
+      ctx.cache.getData('treatments').filter(function (t) { return t._id === ghost._id; }).should.have.length(0);
+      done();
+    });
+  });
 });