Merge pull request #75 from warioishere/codex/update-savecurrent-to-increment-count

Fix pool and client rejected stats flush increments

Author: warioishere

README.md

@@ -2,7 +2,7 @@
 
 Welcome to **BlitzPool**, a lightweight and open-source Bitcoin mining pool based on the [public-pool](https://github.com/benjamin-wilson/public-pool) project – extended with powerful new features and real-world integrations.
 
-Current Version: **v1.3.2**
+Current Version: **v1.3.3**
 
 🌐 **Live Pool:** [https://blitzpool.yourdevice.ch/#/](https://blitzpool.yourdevice.ch/#/)
 

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "public-pool",
-  "version": "1.3.2",
+  "version": "1.3.3",
   "description": "",
   "author": "",
   "private": true,

package.json

@@ -0,0 +1,69 @@
+import { DataSource } from 'typeorm';
+
+import { ClientRejectedStatisticsEntity } from './client-rejected-statistics.entity';
+import { ClientRejectedStatisticsService } from './client-rejected-statistics.service';
+
+describe('ClientRejectedStatisticsService', () => {
+  let dataSource: DataSource;
+
+  beforeEach(async () => {
+    dataSource = new DataSource({
+      type: 'sqlite',
+      database: ':memory:',
+      entities: [ClientRejectedStatisticsEntity],
+      synchronize: true,
+    });
+    await dataSource.initialize();
+  });
+
+  afterEach(async () => {
+    if (dataSource?.isInitialized) {
+      await dataSource.destroy();
+    }
+  });
+
+  it('accumulates deltas when flushing over an existing row', async () => {
+    const repository = dataSource.getRepository(ClientRejectedStatisticsEntity);
+    const service = new ClientRejectedStatisticsService(repository as any);
+
+    const tenMinutes = 1000 * 60 * 10;
+    const now = 1700000000000;
+    const timeSlot = Math.floor(now / tenMinutes) * tenMinutes;
+
+    await repository.insert({
+      address: 'addr',
+      reason: 'duplicate',
+      time: timeSlot,
+      count: 10,
+      shares: 100,
+    });
+
+    const nowSpy = jest.spyOn(Date, 'now').mockReturnValue(timeSlot + 5_000);
+
+    try {
+      await service.addRejectedShare('addr', 'duplicate', 5);
+      await (service as any).saveCurrent();
+
+      let row = await repository.findOneBy({
+        address: 'addr',
+        reason: 'duplicate',
+        time: timeSlot,
+      });
+
+      expect(row).toMatchObject({ count: 11, shares: 104 });
+
+      await service.addRejectedShare('addr', 'duplicate', 3);
+      await (service as any).saveCurrent();
+
+      row = await repository.findOneBy({
+        address: 'addr',
+        reason: 'duplicate',
+        time: timeSlot,
+      });
+
+      expect(row).toMatchObject({ count: 12, shares: 106 });
+    } finally {
+      nowSpy.mockRestore();
+    }
+  });
+});

src/ORM/client-rejected-statistics/client-rejected-statistics.service.spec.ts

@@ -36,32 +36,14 @@ export class ClientRejectedStatisticsService {
 
       if (this.currentTimeSlot == null) {
         this.currentTimeSlot = timeSlot;
-        const existing = await this.clientRejectedStatisticsRepository.findBy({ time: timeSlot });
         this.counts.clear();
-        for (const rec of existing) {
-          if (!this.counts.has(rec.address)) {
-            this.counts.set(rec.address, new Map());
-          }
-          this.counts
-            .get(rec.address)
-            .set(rec.reason, { count: rec.count, shares: rec.shares ?? 0 });
-        }
         this.lastSave = now;
       }
 
       if (this.currentTimeSlot !== timeSlot) {
         await this.saveCurrent();
         this.currentTimeSlot = timeSlot;
-        const existing = await this.clientRejectedStatisticsRepository.findBy({ time: timeSlot });
         this.counts.clear();
-        for (const rec of existing) {
-          if (!this.counts.has(rec.address)) {
-            this.counts.set(rec.address, new Map());
-          }
-          this.counts
-            .get(rec.address)
-            .set(rec.reason, { count: rec.count, shares: rec.shares ?? 0 });
-        }
         this.lastSave = now;
       }
 
@@ -83,29 +65,51 @@ export class ClientRejectedStatisticsService {
   }
 
   private async saveCurrent() {
-    for (const [address, reasons] of this.counts) {
-      for (const [reason, stats] of reasons) {
-        const existing = await this.clientRejectedStatisticsRepository.findOneBy({
+    if (this.counts.size === 0) {
+      return;
+    }
+
+    const values: Array<{
+      time: number;
+      address: string;
+      reason: string;
+      count: number;
+      shares: number;
+    }> = [];
+
+    for (const [address, reasons] of this.counts.entries()) {
+      for (const [reason, stats] of reasons.entries()) {
+        if (stats.count === 0 && stats.shares === 0) {
+          continue;
+        }
+
+        values.push({
           time: this.currentTimeSlot,
           address,
           reason,
+          count: stats.count,
+          shares: stats.shares,
         });
-        if (existing) {
-          await this.clientRejectedStatisticsRepository.update(
-            { time: this.currentTimeSlot, address, reason },
-            { count: stats.count, shares: stats.shares, updatedAt: new Date() },
-          );
-        } else {
-          await this.clientRejectedStatisticsRepository.insert({
-            time: this.currentTimeSlot,
-            address,
-            reason,
-            count: stats.count,
-            shares: stats.shares,
-          });
-        }
       }
     }
+
+    if (values.length === 0) {
+      this.counts.clear();
+      return;
+    }
+
+    await this.clientRejectedStatisticsRepository
+      .createQueryBuilder()
+      .insert()
+      .into(ClientRejectedStatisticsEntity)
+      .values(values)
+      .onConflict(
+        '("time", "address", "reason") DO UPDATE SET "count" = "count" + EXCLUDED."count", "shares" = COALESCE("shares", 0) + EXCLUDED."shares", "updatedAt" = :updatedAt',
+      )
+      .setParameters({ updatedAt: new Date() })
+      .execute();
+
+    this.counts.clear();
   }
 
   public async getTotalsSince(

src/ORM/client-rejected-statistics/client-rejected-statistics.service.ts

@@ -86,22 +86,14 @@ export class PoolRejectedStatisticsService {
 
       if (this.currentTimeSlot == null) {
         this.currentTimeSlot = timeSlot;
-        const existing = await this.poolRejectedStatisticsRepository.findBy({ time: timeSlot });
         this.counts.clear();
-        for (const rec of existing) {
-          this.counts.set(rec.reason, rec.count);
-        }
         this.lastSave = now;
       }
 
       if (this.currentTimeSlot !== timeSlot) {
         await this.saveCurrent();
         this.currentTimeSlot = timeSlot;
-        const existing = await this.poolRejectedStatisticsRepository.findBy({ time: timeSlot });
         this.counts.clear();
-        for (const rec of existing) {
-          this.counts.set(rec.reason, rec.count);
-        }
         this.lastSave = now;
       }
 
@@ -140,17 +132,35 @@ export class PoolRejectedStatisticsService {
   }
 
   private async saveCurrent() {
-    for (const [reason, count] of this.counts) {
-      const existing = await this.poolRejectedStatisticsRepository.findOneBy({ time: this.currentTimeSlot, reason });
-      if (existing) {
-        await this.poolRejectedStatisticsRepository.update(
-          { time: this.currentTimeSlot, reason },
-          { count, updatedAt: new Date() },
-        );
-      } else {
-        await this.poolRejectedStatisticsRepository.insert({ time: this.currentTimeSlot, reason, count });
-      }
+    if (this.counts.size === 0) {
+      return;
     }
+
+    const values = Array.from(this.counts.entries())
+      .filter(([, delta]) => delta !== 0)
+      .map(([reason, delta]) => ({
+        time: this.currentTimeSlot,
+        reason,
+        count: delta,
+      }));
+
+    if (values.length === 0) {
+      this.counts.clear();
+      return;
+    }
+
+    await this.poolRejectedStatisticsRepository
+      .createQueryBuilder()
+      .insert()
+      .into(PoolRejectedStatisticsEntity)
+      .values(values)
+      .onConflict(
+        '("time", "reason") DO UPDATE SET "count" = "count" + EXCLUDED."count", "updatedAt" = :updatedAt',
+      )
+      .setParameters({ updatedAt: new Date() })
+      .execute();
+
+    this.counts.clear();
   }
 
   public async getTotalsSince(time: number): Promise<Record<string, number>> {