share allocation fix

src/pool/index.ts

@@ -48,9 +48,11 @@ export default class Pool {
     let totalWork = 0;
     const walletHashrateMap = new Map<string, number>();
 
-    // Process contributions within the sliding window
-    for (const contribution of this.sharesManager.dumpContributions(10000)) { // 10 seconds window
-      const { address, difficulty, minerId } = contribution;
+    // Get all shares since the last allocation
+    const shares = this.sharesManager.getSharesSinceLastAllocation();
+
+    for (const share of shares) {
+      const { address, difficulty, minerId } = share;
 
       // Aggregate work by address
       if (!works.has(address)) {
@@ -94,7 +96,7 @@ export default class Pool {
       await this.database.addBalance(work.minerId, address, share);
 
       // Track rewards for the miner
-      this.pushMetrics.updateMinerRewardGauge(address, work.minerId, 'block_hash_placeholder'); // Replace 'block_hash_placeholder' with the actual block hash
+      this.pushMetrics.updateMinerRewardGauge(address, work.minerId, 'block_hash_placeholder');
 
       if (DEBUG) {
         this.monitoring.debug(`Pool: Reward of ${sompiToKaspaStringWithSuffix(share, this.treasury.processor.networkId!)} was ALLOCATED to ${work.minerId} with difficulty ${work.difficulty}`);
@@ -104,4 +106,4 @@ export default class Pool {
     // Handle pool fee revenue
     if (works.size > 0) this.revenuize(poolFee);
   }
-}
+}

src/stratum/sharesManager.ts

@@ -53,13 +53,17 @@ export class SharesManager {
   private poolAddress: string;
   private pushGateway: Pushgateway<RegistryContentType>;
   private monitoring: Monitoring;
+  private shareWindow: Denque<Contribution>;
+  private lastAllocationTime: number;
 
   constructor(poolAddress: string, pushGatewayUrl: string) {
     this.poolAddress = poolAddress;
     this.monitoring = new Monitoring();
     this.pushGateway = new Pushgateway<RegistryContentType>(pushGatewayUrl);
     this.startHashRateLogging(60000);
     this.startStatsThread(); // Start the stats logging thread
+    this.shareWindow = new Denque();
+    this.lastAllocationTime = Date.now();
   }
 
   getOrCreateWorkerStats(workerName: string, minerData: MinerData): WorkerStats {
@@ -169,6 +173,9 @@ export class SharesManager {
     }
 
     if (DEBUG) this.monitoring.debug(`SharesManager: Contributed block added from: ${minerId} with address ${address} for nonce: ${nonce}`);
+
+    const share = { minerId, address, difficulty, timestamp: Date.now() };
+    this.shareWindow.push(share);
   }
 
   startStatsThread() {
@@ -333,4 +340,14 @@ export class SharesManager {
       });
     }
   }
+
+  getSharesSinceLastAllocation(): Contribution[] {
+    const currentTime = Date.now();
+    const shares = [];
+    while (this.shareWindow.length > 0 && (this.shareWindow.peekFront()?.timestamp ?? 0) >= this.lastAllocationTime) {
+      shares.push(this.shareWindow.shift()!);
+    }
+    this.lastAllocationTime = currentTime;
+    return shares;
+  }
 }