Revert "improve stratum connection and broadcast performance"

This reverts commit d6d3727119.

src/models/StratumV1Client.spec.ts

@@ -131,9 +131,6 @@ describe('StratumV1Client', () => {
         });
         (StratumV1Client as any).blockedUserAgentLogState.clear();
         (StratumV1Client as any).validationErrorLogState.clear();
-        (StratumV1Client as any).jobBroadcastQueue = [];
-        (StratumV1Client as any).jobBroadcastDraining = false;
-        (StratumV1Client as any).jobBroadcastDrainOffset = 0;
 
         bitcoinRpcService = {
             newBlockTemplate$: newBlockEmitter.asObservable(),
@@ -302,41 +299,6 @@ describe('StratumV1Client', () => {
         await secondClient.destroy();
     });
 
-    it('should close clients that do not complete the stratum handshake', async () => {
-        (configService.get as jest.Mock).mockImplementation((key: string) => {
-            switch (key) {
-                case 'STRATUM_HANDSHAKE_TIMEOUT_MS':
-                    return '1000';
-                case 'DEV_FEE_ADDRESS':
-                    return 'tb1qumezefzdeqqwn5zfvgdrhxjzc5ylr39uhuxcz4';
-                case 'NETWORK':
-                    return 'testnet';
-            }
-            return null;
-        });
-
-        const slowSocket = new Socket();
-        jest.spyOn(slowSocket, 'on').mockImplementation(() => slowSocket);
-        jest.spyOn(slowSocket, 'destroy').mockImplementation(() => slowSocket);
-        slowSocket.end = jest.fn();
-        const slowClient = new StratumV1Client(
-            slowSocket,
-            stratumV1JobsService,
-            bitcoinRpcService,
-            clientService,
-            clientStatisticsService,
-            notificationService,
-            blocksService,
-            configService,
-            moduleRef.get<AddressSettingsService>(AddressSettingsService)
-        );
-
-        jest.advanceTimersByTime(1000);
-
-        expect(slowSocket.destroy).toHaveBeenCalled();
-        await slowClient.destroy();
-    });
-
 
     it('should respond to mining.configure', async () => {
 
@@ -432,32 +394,6 @@ describe('StratumV1Client', () => {
 
     });
 
-    it('should queue mining job broadcasts instead of building jobs in the observable callback', async () => {
-        jest.useFakeTimers();
-        jest.setSystemTime(new Date(parseInt(MockRecording1.TIME, 16) * 1000));
-        jest.spyOn(client as any, 'write').mockImplementation((data) => Promise.resolve(true));
-        const enqueueNewMiningJobSpy = jest.spyOn(client as any, 'enqueueNewMiningJob');
-        const sendNewMiningJobSpy = jest.spyOn(client as any, 'sendNewMiningJob');
-
-        (client as any).clientSubscription = { userAgent: 'bitaxe' };
-        (client as any).clientAuthorization = {
-            address: 'tb1qumezefzdeqqwn5zfvgdrhxjzc5ylr39uhuxcz4',
-            worker: 'bitaxe3'
-        };
-        (client as any).statistics = {};
-        await (client as any).initStratum();
-        await Promise.resolve();
-        await Promise.resolve();
-
-        expect(enqueueNewMiningJobSpy).toHaveBeenCalled();
-
-        jest.advanceTimersByTime(0);
-        await Promise.resolve();
-        await Promise.resolve();
-
-        expect(sendNewMiningJobSpy).toHaveBeenCalled();
-    });
-
     it('should use the header-only fast path for non-block submissions', async () => {
         jest.spyOn(client as any, 'write').mockImplementation((data) => Promise.resolve(true));
         const buildHeaderSpy = jest.spyOn(MiningJob.prototype, 'buildHeaderBuffer');

src/models/StratumV1Client.ts

@@ -31,14 +31,6 @@ import { StratumV1ClientStatistics } from './StratumV1ClientStatistics';
 const TRUE_DIFF_ONE = 2.695953529101131e67;
 const BLOCKED_USER_AGENT_LOG_INTERVAL_MS = 60 * 1000;
 const VALIDATION_ERROR_LOG_INTERVAL_MS = 60 * 1000;
-const DEFAULT_JOB_BROADCAST_BATCH_SIZE = 250;
-const DEFAULT_HANDSHAKE_TIMEOUT_MS = 30000;
-
-interface JobBroadcastQueueItem {
-    client: StratumV1Client;
-    jobTemplate: IJobTemplate;
-    jobTemplateId: string;
-}
 
 export function effectiveJobDifficulty(
     jobIdInt: number,
@@ -59,9 +51,6 @@ export function effectiveJobDifficulty(
 export class StratumV1Client {
     private static blockedUserAgentLogState = new Map<string, { nextLogAt: number, suppressed: number }>();
     private static validationErrorLogState = new Map<string, { nextLogAt: number, suppressed: number, sample: string }>();
-    private static jobBroadcastQueue: JobBroadcastQueueItem[] = [];
-    private static jobBroadcastDraining = false;
-    private static jobBroadcastDrainOffset = 0;
 
     public clientSubscription: SubscriptionMessage;
     private clientConfiguration: ConfigurationMessage;
@@ -87,9 +76,6 @@ export class StratumV1Client {
 
     private buffer: string = '';
     private connectionClosed = false;
-    private destroyed = false;
-    private handshakeTimeout: NodeJS.Timeout | null;
-    private latestQueuedJobTemplateId: string;
 
     private miningSubmissionHashes = new Set<string>()
 
@@ -105,12 +91,6 @@ export class StratumV1Client {
         private readonly addressSettingsService: AddressSettingsService
     ) {
 
-        this.handshakeTimeout = setTimeout(() => {
-            if (!this.stratumInitialized) {
-                this.closeSocket();
-            }
-        }, this.getHandshakeTimeoutMs());
-
         this.socket.on('data', (data: Buffer) => {
             this.buffer += data.toString();
             let lines = this.buffer.split('\n');
@@ -135,12 +115,6 @@ export class StratumV1Client {
     }
 
     public async destroy() {
-        this.connectionClosed = true;
-
-        if (this.destroyed) {
-            return;
-        }
-        this.destroyed = true;
 
         if (this.clientEntity?.id) {
             await this.clientService.delete(this.clientEntity.id);
@@ -150,11 +124,6 @@ export class StratumV1Client {
             this.stratumSubscription.unsubscribe();
         }
 
-        if (this.handshakeTimeout != null) {
-            clearTimeout(this.handshakeTimeout);
-            this.handshakeTimeout = null;
-        }
-
         this.backgroundWork.forEach(work => {
             clearInterval(work);
         });
@@ -416,10 +385,6 @@ export class StratumV1Client {
 
     private async initStratum() {
         this.stratumInitialized = true;
-        if (this.handshakeTimeout != null) {
-            clearTimeout(this.handshakeTimeout);
-            this.handshakeTimeout = null;
-        }
 
         if (this.isBlockedUserAgent(this.clientSubscription.userAgent)) {
             this.logBlockedUserAgent(this.clientSubscription.userAgent);
@@ -448,7 +413,7 @@ export class StratumV1Client {
                 if(jobTemplate.blockData.clearJobs){
                     this.miningSubmissionHashes.clear();
                 }
-                this.enqueueNewMiningJob(jobTemplate);
+                await this.sendNewMiningJob(jobTemplate);
             } catch (e) {
                 await this.socket.end();
                 console.error(e);
@@ -468,68 +433,6 @@ export class StratumV1Client {
         // );
     }
 
-    private enqueueNewMiningJob(jobTemplate: IJobTemplate) {
-        if (this.connectionClosed || this.socket.destroyed || this.socket.writableEnded) {
-            return;
-        }
-
-        this.latestQueuedJobTemplateId = jobTemplate.blockData.id;
-        StratumV1Client.jobBroadcastQueue.push({
-            client: this,
-            jobTemplate,
-            jobTemplateId: jobTemplate.blockData.id
-        });
-        StratumV1Client.scheduleJobBroadcastDrain();
-    }
-
-    private static scheduleJobBroadcastDrain() {
-        if (StratumV1Client.jobBroadcastDraining) {
-            return;
-        }
-
-        StratumV1Client.jobBroadcastDraining = true;
-        setTimeout(() => StratumV1Client.drainJobBroadcastQueue(), 0);
-    }
-
-    private static drainJobBroadcastQueue() {
-        const batchSize = StratumV1Client.getJobBroadcastBatchSize();
-        let processed = 0;
-
-        while (processed < batchSize && StratumV1Client.jobBroadcastDrainOffset < StratumV1Client.jobBroadcastQueue.length) {
-            const item = StratumV1Client.jobBroadcastQueue[StratumV1Client.jobBroadcastDrainOffset++];
-            processed++;
-
-            if (item.client.connectionClosed
-                || item.client.socket.destroyed
-                || item.client.socket.writableEnded
-                || item.client.latestQueuedJobTemplateId !== item.jobTemplateId) {
-                continue;
-            }
-
-            item.client.sendNewMiningJob(item.jobTemplate).catch(async (e) => {
-                item.client.closeSocket();
-                console.error(e);
-            });
-        }
-
-        if (StratumV1Client.jobBroadcastDrainOffset >= StratumV1Client.jobBroadcastQueue.length) {
-            StratumV1Client.jobBroadcastQueue = [];
-            StratumV1Client.jobBroadcastDrainOffset = 0;
-            StratumV1Client.jobBroadcastDraining = false;
-            return;
-        }
-
-        setTimeout(() => StratumV1Client.drainJobBroadcastQueue(), 0);
-    }
-
-    private static getJobBroadcastBatchSize() {
-        const configuredBatchSize = parseInt(process.env.STRATUM_JOB_BROADCAST_BATCH_SIZE, 10);
-        if (Number.isFinite(configuredBatchSize) && configuredBatchSize > 0) {
-            return configuredBatchSize;
-        }
-        return DEFAULT_JOB_BROADCAST_BATCH_SIZE;
-    }
-
     private async sendNewMiningJob(jobTemplate: IJobTemplate) {
 
         let payoutInformation= [
@@ -911,20 +814,8 @@ export class StratumV1Client {
         return ` sample=${values.join(',')}`;
     }
 
-    private getHandshakeTimeoutMs() {
-        const configuredTimeout = parseInt(this.configService.get<string>('STRATUM_HANDSHAKE_TIMEOUT_MS') ?? process.env.STRATUM_HANDSHAKE_TIMEOUT_MS, 10);
-        if (Number.isFinite(configuredTimeout) && configuredTimeout > 0) {
-            return configuredTimeout;
-        }
-        return DEFAULT_HANDSHAKE_TIMEOUT_MS;
-    }
-
     private closeSocket() {
         this.connectionClosed = true;
-        if (this.handshakeTimeout != null) {
-            clearTimeout(this.handshakeTimeout);
-            this.handshakeTimeout = null;
-        }
         if (!this.socket.destroyed) {
             this.socket.destroy();
         }

src/services/stratum-v1.service.ts

@@ -70,8 +70,6 @@ export class StratumV1Service implements OnModuleInit {
 
     private startSocketServer(port: number) {
         const server = new Server(async (socket: Socket) => {
-            socket.setNoDelay(true);
-            socket.setKeepAlive(true, 60000);
             // Set 15-minute timeout
             socket.setTimeout(1000 * 60 * 15);
 
@@ -88,17 +86,9 @@ export class StratumV1Service implements OnModuleInit {
             );
 
             // Unified cleanup function
-            let cleanedUp = false;
             const cleanup = async (reason: string) => {
-                if (cleanedUp) {
+                if (client.extraNonceAndSessionId != null) {
-                    return;
+                    await client.destroy();
-                }
-                cleanedUp = true;
-
-                const initializedClient = client.extraNonceAndSessionId != null;
-                await client.destroy();
-
-                if (initializedClient) {
                     if (reason == 'Error') {
                         this.errorClosure++;
                     } else {
@@ -159,8 +149,6 @@ export class StratumV1Service implements OnModuleInit {
         };
 
         const server = createServer(tlsOptions, async (socket: TLSSocket) => {
-            socket.setNoDelay(true);
-            socket.setKeepAlive(true, 60000);
             // Set 15-minute timeout
             socket.setTimeout(1000 * 60 * 15);
 
@@ -176,17 +164,9 @@ export class StratumV1Service implements OnModuleInit {
                 this.addressSettingsService
             );
 
-            let cleanedUp = false;
             const cleanup = async (reason: string) => {
-                if (cleanedUp) {
+                if (client.extraNonceAndSessionId != null) {
-                    return;
+                    await client.destroy();
-                }
-                cleanedUp = true;
-
-                const initializedClient = client.extraNonceAndSessionId != null;
-                await client.destroy();
-
-                if (initializedClient) {
                     if (reason === 'Error') {
                         this.errorClosure++;
                     } else {