Always flush the buffer on stop, and let unload finish (#577)

README.md

@@ -621,6 +621,8 @@ sendTo('sql.0', 'getEnabledDPs', {}, function (result) {
 
 ## Changelog
 ### **WORK IN PROGRESS**
+* (@GermanBluefox) Fixed buffered values being lost when the adapter stops, and unload no longer hangs when there is nothing else to write (#577)
+* (@GermanBluefox) `getCounter` no longer reads the NULL boundary marker that `writeNulls` stores on start and stop as a zero counter value (#577)
 * (@GermanBluefox) Fixed "Test connection" reporting a failure for a working configuration: the request was built as a text template that could produce invalid JSON, and it left out `dbname`, `doNotCreateDatabase` and the docker settings (#355)
 
 ### **WORK IN PROGRESS**

package.json

@@ -28,7 +28,7 @@
   },
   "dependencies": {
     "@iobroker/adapter-core": "^3.4.3",
-    "@iobroker/aggregate": "^1.0.2",
+    "@iobroker/aggregate": "^1.0.3",
     "@iobroker/plugin-docker": "^1.2.1"
   },
   "devDependencies": {

src/main.ts

@@ -1298,8 +1298,12 @@ export class SqlAdapter extends Adapter {
     }
 
     finish(callback: () => void): void {
-        let count = 0;
         const now = Date.now();
+        // How many datapoints still have to report in. `allFinished()` used to be reached through a
+        // counter shared by every write of every datapoint, which fired it as soon as any one of
+        // them happened to finish last - see #577.
+        let datapointsPending = 0;
+        let allDispatched = false;
 
         const allFinished = (): void => {
             if (this.clientPool) {
@@ -1322,9 +1326,30 @@ export class SqlAdapter extends Adapter {
             }
         };
 
+        const datapointDone = (): void => {
+            datapointsPending--;
+            if (allDispatched && datapointsPending <= 0) {
+                allFinished();
+            }
+        };
+
+        /**
+         * Write out everything one datapoint still holds, then report it done.
+         *
+         * The buffered values are flushed unconditionally. They used to be written only from inside
+         * the `skipped` and `writeNulls` branches, so a datapoint with neither - every SQLite
+         * instance, since writeNulls is forced off there - lost its buffer (#577).
+         *
+         * The counter is per datapoint. It used to be shared across all of them, so the flush ran on
+         * whichever callback happened to bring the shared count to zero and only that one datapoint's
+         * buffer was written; the same counter also let `allFinished()` close the pool while later
+         * datapoints were still being dispatched.
+         *
+         * @param id the datapoint to finish
+         */
         const finishId = (id: string): void => {
             if (!this.sqlDPs[id]) {
-                return;
+                return datapointDone();
             }
             if (this.sqlDPs[id].relogTimeout) {
                 clearTimeout(this.sqlDPs[id].relogTimeout);
@@ -1336,18 +1361,34 @@ export class SqlAdapter extends Adapter {
             }
             const state: IobDataEntryEx | null = this.sqlDPs[id].state ? { ...this.sqlDPs[id].state } : null;
 
+            // Values this datapoint has to write before its buffer goes out, so that the boundary
+            // markers keep their order relative to the buffered values.
+            let pendingWrites = 0;
+            const flushBuffer = (): void => {
+                if (pendingWrites) {
+                    return;
+                }
+                const buffered = this.sqlDPs[id]?.list;
+                // Not every entry in sqlDPs is a configured datapoint: storeState, getHistory and
+                // the object view create stubs with `sqlDPs[id] ||= {}` just to carry an index, and
+                // those have no buffer. The datapoint can also disappear while the writes above are
+                // still in flight.
+                if (!buffered?.length) {
+                    return datapointDone();
+                }
+                this.pushValuesIntoDB(id, buffered, () => datapointDone());
+            };
+            const writeDone = (): void => {
+                pendingWrites--;
+                flushBuffer();
+            };
+
             if (
                 this.sqlDPs[id].skipped &&
                 !(this.sqlDPs[id].config && this.sqlDPs[id].config.disableSkippedValueLogging)
             ) {
-                count++;
-                this.pushValueIntoDB(id, this.sqlDPs[id].skipped, false, true, () => {
-                    if (!--count) {
-                        this.pushValuesIntoDB(id, this.sqlDPs[id].list, () => {
-                            allFinished();
-                        });
-                    }
-                });
+                pendingWrites++;
+                this.pushValueIntoDB(id, this.sqlDPs[id].skipped, false, true, writeDone);
                 this.sqlDPs[id].skipped = null;
             }
 
@@ -1361,39 +1402,28 @@ export class SqlAdapter extends Adapter {
             };
 
             if (this.sqlDPs[id].config && this.config.writeNulls) {
+                pendingWrites++;
                 if (this.sqlDPs[id].config.changesOnly && state && state.val !== null) {
-                    count++;
-                    ((_id: string, _state: IobDataEntryEx, _nullValue: IobDataEntryEx): void => {
-                        _state.ts = now;
-                        _state.from = `system.adapter.${this.namespace}`;
-                        nullValue.ts += 4;
-                        nullValue.lc += 4; // because of MS SQL
-                        this.log.debug(`Write 1/2 "${_state.val}" _id: ${_id}`);
-                        this.pushValueIntoDB(_id, _state, false, true, () => {
-                            // terminate values with null to indicate adapter stop. timestamp + 1
-                            this.log.debug(`Write 2/2 "null" _id: ${_id}`);
-                            this.pushValueIntoDB(_id, _nullValue, false, true, () => {
-                                if (!--count) {
-                                    this.pushValuesIntoDB(id, this.sqlDPs[id].list, () => {
-                                        allFinished();
-                                    });
-                                }
-                            });
-                        });
-                    })(id, state, nullValue);
+                    state.ts = now;
+                    state.from = `system.adapter.${this.namespace}`;
+                    nullValue.ts += 4;
+                    nullValue.lc += 4; // because of MS SQL
+                    this.log.debug(`Write 1/2 "${state.val}" _id: ${id}`);
+                    this.pushValueIntoDB(id, state, false, true, () => {
+                        // terminate values with null to indicate adapter stop. timestamp + 1
+                        this.log.debug(`Write 2/2 "null" _id: ${id}`);
+                        this.pushValueIntoDB(id, nullValue, false, true, writeDone);
+                    });
                 } else {
                     // terminate values with null to indicate adapter stop. timestamp + 1
-                    count++;
                     this.log.debug(`Write 0 NULL _id: ${id}`);
-                    this.pushValueIntoDB(id, nullValue, false, true, () => {
-                        if (!--count) {
-                            this.pushValuesIntoDB(id, this.sqlDPs[id].list, () => {
-                                allFinished();
-                            });
-                        }
-                    });
+                    this.pushValueIntoDB(id, nullValue, false, true, writeDone);
                 }
             }
+
+            // Runs straight away when this datapoint had nothing else to write, otherwise the last
+            // writeDone() gets here.
+            flushBuffer();
         };
 
         if (!this.subscribeAll) {
@@ -1445,9 +1475,13 @@ export class SqlAdapter extends Adapter {
                 continue;
             }
             dpcount++;
+            datapointsPending++;
             delay += dpcount % 50 === 0 ? 1000 : 0;
             setTimeout(finishId, delay, id);
         }
+        // Only now may a datapoint reporting in finish the whole run: the dispatch above is spread
+        // over seconds, so an early finisher must not close the pool under the ones still queued.
+        allDispatched = true;
 
         if (!dpcount && callback) {
             if (this.clientPool) {

test/testCommons.js

@@ -133,6 +133,33 @@ describe('Test Common functions', function () {
         commons.sendResponseCounter(adapter, {}, { start: 5, end: 95 }, timeSeries);
     });
 
+    // writeNulls puts a NULL boundary marker into ts_number whenever the adapter starts or stops, and
+    // getCounterDiff's "last row before the window" subquery has no `val IS NOT NULL` filter - so that
+    // marker is the first entry getCounter sees. Treating it as the number 0 interpolates the value at
+    // the window start somewhere between 0 and the first real value, inflating the total (#577). The
+    // marker carries no counter reading and has to be dropped instead.
+    it('Test Common functions: counter ignores a NULL boundary marker before the window', function (done) {
+        const timeSeries = [
+            { ts: 0, val: null },
+            { ts: 1000, val: 100 },
+            { ts: 2000, val: 200 },
+            { ts: 3000, val: 10 },
+            { ts: 4000, val: 110 },
+            { ts: 6000, val: null },
+        ];
+
+        const adapter = {
+            sendTo: function (from, command, result, callback) {
+                // (200 - 100) + (110 - 10); reading the NULL as 0 yields 210
+                assert.strictEqual(result.result, 200);
+                done();
+            },
+            log,
+        };
+
+        commons.sendResponseCounter(adapter, {}, { start: 900, end: 5000 }, timeSeries);
+    });
+
     // "onchange" ("raw" in the e-charts UI) must be passed through untouched. aggregationLogic() has no
     // branch for it, so running it through the bucket aggregation returns one entry per interval with
     // val === null - i.e. an empty chart (issue #522).