feat: implement last data send tracking for order guarantee

takaokouji · Gemini · takaokouji · commit 080a72b3b2a2 · 2026-01-03T12:15:48.000+09:00
- Added lastDataSendPromise to track the most recent data transmission completion - Updated fireEventsBatch to await lastDataSendPromise before sending events - Updated handleDataUpdate and fetchAllNodesData to use server timestamps - Added unit test for data and event ordering (mesh_service_v2_order.js) - Updated mesh_service_v2_timestamp.js to match new server timestamp behavior Related to smalruby/smalruby3-gui#500 🤖 Generated with [Gemini Code](https://gemini.google.com/code) Co-Authored-By: Gemini <noreply@google.com>
diff --git a/src/extensions/scratch3_mesh_v2/mesh-service.js b/src/extensions/scratch3_mesh_v2/mesh-service.js
@@ -48,6 +48,10 @@ class MeshV2Service {
         this.connectionTimer = null;
         this.heartbeatTimer = null;
         this.dataSyncTimer = null;
+
+        // Last data send promise to track completion of the most recent data transmission
+        this.lastDataSendPromise = Promise.resolve();
+
         this.memberHeartbeatInterval = 120; // Default 2 min
 
         // Data from other nodes: { nodeId: { key: { value: string, timestamp: number } } }
@@ -439,10 +443,13 @@ class MeshV2Service {
             this.remoteData[nodeId] = {};
         }
 
+        // Use server timestamp with fallback to current time
+        const serverTimestamp = nodeStatus.timestamp ? new Date(nodeStatus.timestamp).getTime() : Date.now();
+
         nodeStatus.data.forEach(item => {
             this.remoteData[nodeId][item.key] = {
                 value: item.value,
-                timestamp: Date.now() // Add timestamp
+                timestamp: serverTimestamp
             };
         });
     }
@@ -617,8 +624,8 @@ class MeshV2Service {
         if (!this.groupId || !this.client || events.length === 0) return;
 
         try {
-            // データ送信完了を待つ (Removed to prevent blocking, relying on unified subscription for order)
-            // await this.dataRateLimiter.waitForCompletion();
+            // Wait for last data send to complete
+            await this.lastDataSendPromise;
 
             this.costTracking.mutationCount++;
             this.costTracking.fireEventsCount++;
@@ -778,7 +785,9 @@ class MeshV2Service {
         }
 
         try {
-            await this.dataRateLimiter.send(dataArray, this._reportDataBound);
+            // Save Promise to track completion (including queue time)
+            this.lastDataSendPromise = this.dataRateLimiter.send(dataArray, this._reportDataBound);
+            await this.lastDataSendPromise;
         } catch (error) {
             log.error(`Mesh V2: Failed to send data: ${error}`);
             const reason = this.shouldDisconnectOnError(error);
@@ -792,25 +801,43 @@ class MeshV2Service {
      * Internal method to send data to the server.
      * Used as sendFunction in dataRateLimiter.
      * @param {Array} payload - Array of {key, value} objects.
+     * @returns {Promise} - Resolves with the mutation result.
      * @private
      */
     async _reportData (payload) {
-        this.costTracking.mutationCount++;
-        this.costTracking.reportDataCount++;
-        await this.client.mutate({
-            mutation: REPORT_DATA,
-            variables: {
-                groupId: this.groupId,
-                domain: this.domain,
-                nodeId: this.meshId,
-                data: payload
-            }
-        });
+        if (!this.groupId || !this.client) return;
 
-        // Update last sent data on success
-        payload.forEach(item => {
-            this.lastSentData[item.key] = item.value;
-        });
+        try {
+            this.costTracking.mutationCount++;
+            this.costTracking.reportDataCount++;
+
+            // Save Promise to track completion
+            this.lastDataSendPromise = this.client.mutate({
+                mutation: REPORT_DATA,
+                variables: {
+                    groupId: this.groupId,
+                    domain: this.domain,
+                    nodeId: this.meshId,
+                    data: payload
+                }
+            });
+
+            const result = await this.lastDataSendPromise;
+
+            // Update last sent data on success
+            payload.forEach(item => {
+                this.lastSentData[item.key] = item.value;
+            });
+
+            return result;
+        } catch (error) {
+            log.error(`Mesh V2: Failed to report data: ${error}`);
+            const reason = this.shouldDisconnectOnError(error);
+            if (reason) {
+                this.cleanupAndDisconnect(reason);
+            }
+            throw error;
+        }
     }
 
     fireEvent (eventName, payload = '') {
@@ -902,10 +929,11 @@ class MeshV2Service {
                 if (!this.remoteData[status.nodeId]) {
                     this.remoteData[status.nodeId] = {};
                 }
+                const serverTimestamp = status.timestamp ? new Date(status.timestamp).getTime() : Date.now();
                 status.data.forEach(item => {
                     this.remoteData[status.nodeId][item.key] = {
                         value: item.value,
-                        timestamp: Date.now()
+                        timestamp: serverTimestamp
                     };
                 });
             });
diff --git a/test/unit/mesh_service_v2_order.js b/test/unit/mesh_service_v2_order.js
@@ -0,0 +1,121 @@
+const test = require('tap').test;
+const MeshV2Service = require('../../src/extensions/scratch3_mesh_v2/mesh-service');
+const {REPORT_DATA, FIRE_EVENTS} = require('../../src/extensions/scratch3_mesh_v2/gql-operations');
+
+const createMockBlocks = () => ({
+    runtime: {
+        sequencer: {},
+        emit: () => {},
+        on: () => {},
+        off: () => {},
+        getTargetForStage: () => ({
+            variables: {}
+        })
+    },
+    opcodeFunctions: {
+        event_broadcast: () => {}
+    }
+});
+
+test('MeshV2Service Data and Event Order', t => {
+    t.test('fireEventsBatch awaits lastDataSendPromise', async st => {
+        const blocks = createMockBlocks();
+        const service = new MeshV2Service(blocks, 'node1', 'domain1');
+        service.stopEventBatchTimer();
+        service.groupId = 'group1';
+
+        let dataMutationStarted = false;
+        let dataMutationFinished = false;
+        let eventMutationStarted = false;
+
+        service.client = {
+            mutate: options => {
+                if (options.mutation === REPORT_DATA) {
+                    dataMutationStarted = true;
+                    return new Promise(resolve => {
+                        setTimeout(() => {
+                            dataMutationFinished = true;
+                            resolve({data: {reportDataByNode: {
+                                nodeId: 'node1',
+                                timestamp: new Date().toISOString(),
+                                data: []
+                            }}});
+                        }, 50); // Delay data mutation
+                    });
+                }
+                if (options.mutation === FIRE_EVENTS) {
+                    eventMutationStarted = true;
+                    st.ok(dataMutationStarted, 'Data mutation should have started');
+                    st.ok(dataMutationFinished, 'Data mutation should have finished before event mutation starts');
+                    return Promise.resolve({data: {fireEventsByNode: {}}});
+                }
+                return Promise.resolve({});
+            }
+        };
+
+        // 1. Send data
+        const dataPromise = service.sendData([{key: 'var1', value: '10'}]);
+        st.ok(service.lastDataSendPromise, 'lastDataSendPromise should be set');
+        
+        // 2. Fire event batch immediately (should wait for dataPromise)
+        const eventPromise = service.fireEventsBatch([{eventName: 'msg1', payload: '', firedAt: 't1'}]);
+
+        await Promise.all([dataPromise, eventPromise]);
+        
+        st.ok(dataMutationFinished, 'Data mutation finished');
+        st.ok(eventMutationStarted, 'Event mutation started');
+
+        st.end();
+    });
+
+    t.test('handleDataUpdate uses server timestamp', st => {
+        const blocks = createMockBlocks();
+        const service = new MeshV2Service(blocks, 'node1', 'domain1');
+        
+        const serverTimestamp = '2025-12-30T12:34:56.789Z';
+        const expectedTime = new Date(serverTimestamp).getTime();
+        
+        const nodeStatus = {
+            nodeId: 'node2',
+            timestamp: serverTimestamp,
+            data: [
+                {key: 'var1', value: '100'}
+            ]
+        };
+        
+        service.handleDataUpdate(nodeStatus);
+        
+        st.equal(service.remoteData.node2.var1.value, '100');
+        st.equal(service.remoteData.node2.var1.timestamp, expectedTime, 'Should use server timestamp');
+        
+        st.end();
+    });
+
+    t.test('fireEventsBatch works without preceding data send', async st => {
+        const blocks = createMockBlocks();
+        const service = new MeshV2Service(blocks, 'node1', 'domain1');
+        service.stopEventBatchTimer();
+        service.groupId = 'group1';
+
+        let eventMutationStarted = false;
+
+        service.client = {
+            mutate: options => {
+                if (options.mutation === FIRE_EVENTS) {
+                    eventMutationStarted = true;
+                    return Promise.resolve({data: {fireEventsByNode: {}}});
+                }
+                return Promise.resolve({});
+            }
+        };
+
+        // fireEventsBatch should work even if lastDataSendPromise is just Promise.resolve()
+        await service.fireEventsBatch([{eventName: 'msg1', payload: '', firedAt: 't1'}]);
+        
+        st.ok(eventMutationStarted, 'Event mutation started without data send');
+
+        st.end();
+    });
+
+    t.end();
+});
diff --git a/test/unit/mesh_service_v2_timestamp.js b/test/unit/mesh_service_v2_timestamp.js
@@ -35,48 +35,48 @@ test('MeshV2Service Timestamp-based getRemoteVariable', t => {
         st.end();
     });
 
-    t.test('handleDataUpdate should add timestamp', st => {
+    t.test('handleDataUpdate should add timestamp from nodeStatus', st => {
+        const serverTimestamp = new Date().toISOString();
+        const expectedTimestamp = new Date(serverTimestamp).getTime();
         const nodeStatus = {
             nodeId: 'node4',
+            timestamp: serverTimestamp,
             data: [
                 {key: 'var1', value: '100'}
             ]
         };
 
-        const beforeUpdate = Date.now();
         service.handleDataUpdate(nodeStatus);
-        const afterUpdate = Date.now();
 
         st.ok(service.remoteData.node4, 'Node 4 should be added');
         st.ok(service.remoteData.node4.var1, 'var1 should be added');
         st.equal(service.remoteData.node4.var1.value, '100');
-        st.ok(service.remoteData.node4.var1.timestamp >= beforeUpdate);
-        st.ok(service.remoteData.node4.var1.timestamp <= afterUpdate);
+        st.equal(service.remoteData.node4.var1.timestamp, expectedTimestamp, 'Should use server timestamp');
         st.end();
     });
 
-    t.test('fetchAllNodesData should add timestamp', async st => {
+    t.test('fetchAllNodesData should add timestamp from status', async st => {
+        const serverTimestamp = new Date().toISOString();
+        const expectedTimestamp = new Date(serverTimestamp).getTime();
         service.client = {
             query: () => Promise.resolve({
                 data: {
                     listGroupStatuses: [
                         {
                             nodeId: 'node5',
+                            timestamp: serverTimestamp,
                             data: [{key: 'var2', value: '200'}]
                         }
                     ]
                 }
             })
         };
 
-        const beforeFetch = Date.now();
         await service.fetchAllNodesData();
-        const afterFetch = Date.now();
 
         st.ok(service.remoteData.node5, 'Node 5 should be added');
         st.equal(service.remoteData.node5.var2.value, '200');
-        st.ok(service.remoteData.node5.var2.timestamp >= beforeFetch);
-        st.ok(service.remoteData.node5.var2.timestamp <= afterFetch);
+        st.equal(service.remoteData.node5.var2.timestamp, expectedTimestamp, 'Should use server timestamp');
         st.end();
     });
 
