Extract SigningService from MessageSigner for clearer separation of concerns

mondoreale · mondoreale · commit 90f46f35164e · 2026-01-29T19:54:15.000+01:00
- SigningService: Container-scoped service that owns the signing worker,
  provides sign() method, and handles worker lifecycle (lazy init, cleanup)
- MessageSigner: Now injects SigningService instead of managing worker directly

This decouples the worker management from message signing logic. Each
StreamrClient gets its own SigningService instance, with the worker
destroyed automatically when the client is destroyed.

Test utilities updated with createMessageSigner() helper that uses a
mock SigningService (no worker) for faster, simpler tests.
diff --git a/packages/sdk/src/signature/MessageSigner.ts b/packages/sdk/src/signature/MessageSigner.ts
@@ -4,24 +4,19 @@ import { Identity, IdentityInjectionToken } from '../identity/Identity'
 import { StreamMessage, StreamMessageOptions } from '../protocol/StreamMessage'
 import { createSignaturePayload } from './createSignaturePayload'
 import { SignatureType } from '@streamr/trackerless-network'
-import { Signing } from './Signing'
-import { DestroySignal } from '../DestroySignal'
+import { SigningService } from './SigningService'
 
 @scoped(Lifecycle.ContainerScoped)
 export class MessageSigner {
     private readonly identity: Identity
-    private signing: Signing | undefined
+    private readonly signingService: SigningService
 
     constructor(
         @inject(IdentityInjectionToken) identity: Identity,
-            destroySignal?: DestroySignal
+            signingService: SigningService
     ) {
         this.identity = identity
-        destroySignal?.onDestroy.listen(() => this.destroy())
-    }
-
-    private getSigning(): Signing {
-        return this.signing ??= new Signing()
+        this.signingService = signingService
     }
 
     async createSignedMessage(
@@ -51,7 +46,7 @@ export class MessageSigner {
         signatureType: SignatureType,
         privateKey: Uint8Array
     ): Promise<Uint8Array> {
-        const result = await this.getSigning().createSignature({
+        const result = await this.signingService.sign({
             payloadInput: opts,
             privateKey,
             signatureType
@@ -63,14 +58,4 @@ export class MessageSigner {
 
         return result.signature
     }
-
-    /**
-     * Cleanup worker resources when the signer is no longer needed.
-     */
-    destroy(): void {
-        if (this.signing) {
-            this.signing.destroy()
-            this.signing = undefined
-        }
-    }
 }
diff --git a/packages/sdk/src/signature/Signing.ts b/packages/sdk/src/signature/Signing.ts
diff --git a/packages/sdk/src/signature/SigningService.ts b/packages/sdk/src/signature/SigningService.ts
@@ -0,0 +1,46 @@
+/**
+ * Singleton signing service using Web Worker.
+ * This offloads CPU-intensive cryptographic operations to a separate thread.
+ * Works in both browser and Node.js environments via platform-specific config.
+ * 
+ * The worker is lazily initialized on first use and shared across all MessageSigner instances.
+ */
+import { wrap, releaseProxy, type Remote } from 'comlink'
+import { Lifecycle, scoped } from 'tsyringe'
+import { createSigningWorker } from '@/createSigningWorker'
+import { SigningResult, SigningRequest } from './signingUtils'
+import type { SigningWorkerApi } from './SigningWorker'
+import { DestroySignal } from '../DestroySignal'
+
+@scoped(Lifecycle.ContainerScoped)
+export class SigningService {
+    private worker: ReturnType<typeof createSigningWorker> | undefined
+    private workerApi: Remote<SigningWorkerApi> | undefined
+
+    constructor(destroySignal: DestroySignal) {
+        destroySignal.onDestroy.listen(() => this.destroy())
+    }
+
+    private getWorkerApi(): Remote<SigningWorkerApi> {
+        if (this.workerApi === undefined) {
+            this.worker = createSigningWorker()
+            this.workerApi = wrap<SigningWorkerApi>(this.worker)
+        }
+        return this.workerApi
+    }
+
+    async sign(request: SigningRequest): Promise<SigningResult> {
+        return this.getWorkerApi().createSignature(request)
+    }
+
+    destroy(): void {
+        if (this.workerApi !== undefined) {
+            this.workerApi[releaseProxy]()
+            this.workerApi = undefined
+        }
+        if (this.worker !== undefined) {
+            this.worker.terminate()
+            this.worker = undefined
+        }
+    }
+}
diff --git a/packages/sdk/test/end-to-end/publish-subscribe-raw.test.ts b/packages/sdk/test/end-to-end/publish-subscribe-raw.test.ts
@@ -1,8 +1,7 @@
 import { StreamID, toUserId } from '@streamr/utils'
-import { createTestClient, createTestStream } from '../test-utils/utils'
+import { createMessageSigner, createTestClient, createTestStream } from '../test-utils/utils'
 import { nextValue } from '../../src/utils/iterators'
 import { EthereumKeyPairIdentity } from '../../src/identity/EthereumKeyPairIdentity'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { Wallet } from 'ethers'
 import { MessageID } from '../../src/protocol/MessageID'
 import { ContentType, EncryptionType, SignatureType } from '@streamr/trackerless-network'
@@ -29,7 +28,7 @@ describe('publish-subscribe-raw', () => {
     })
 
     async function createTestMessage() {
-        const messageSigner = new MessageSigner(EthereumKeyPairIdentity.fromPrivateKey(publisherWallet.privateKey))
+        const messageSigner = createMessageSigner(EthereumKeyPairIdentity.fromPrivateKey(publisherWallet.privateKey))
         return await messageSigner.createSignedMessage({
             messageId: new MessageID(streamId, 0, 123456789, 0, toUserId(publisherWallet.address), 'mock-msgChainId'),
             content: new Uint8Array([1, 2, 3]),
diff --git a/packages/sdk/test/integration/Resends.test.ts b/packages/sdk/test/integration/Resends.test.ts
@@ -6,10 +6,9 @@ import { StreamrClient } from '../../src/StreamrClient'
 import { GroupKey } from '../../src/encryption/GroupKey'
 import { StreamPermission } from '../../src/permission'
 import { MessageFactory } from '../../src/publish/MessageFactory'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
 import { FakeEnvironment } from '../test-utils/fake/FakeEnvironment'
-import { createGroupKeyQueue, createStreamRegistry } from '../test-utils/utils'
+import { createGroupKeyQueue, createMessageSigner, createStreamRegistry } from '../test-utils/utils'
 import { EthereumKeyPairIdentity } from '../../src/identity/EthereumKeyPairIdentity'
 import { createStrictConfig } from '../../src/Config'
 
@@ -45,7 +44,7 @@ describe('Resends', () => {
             streamRegistry: createStreamRegistry(),
             groupKeyQueue: await createGroupKeyQueue(identity, groupKey),
             signatureValidator: mock<SignatureValidator>(),
-            messageSigner: new MessageSigner(identity),
+            messageSigner: createMessageSigner(identity),
             config: createStrictConfig()
         })
         // store the encryption key publisher's local group key store
diff --git a/packages/sdk/test/integration/Subscriber2.test.ts b/packages/sdk/test/integration/Subscriber2.test.ts
@@ -10,7 +10,7 @@ import { MessageSigner } from '../../src/signature/MessageSigner'
 import { Subscription } from '../../src/subscribe/Subscription'
 import { FakeEnvironment } from '../test-utils/fake/FakeEnvironment'
 import { getPublishTestStreamMessages } from '../test-utils/publish'
-import { createTestStream } from '../test-utils/utils'
+import { createMessageSigner, createTestStream } from '../test-utils/utils'
 import { MessageID } from './../../src/protocol/MessageID'
 import { StreamMessage, StreamMessageType } from './../../src/protocol/StreamMessage'
 import { ContentType, EncryptionType, SignatureType } from '@streamr/trackerless-network'
@@ -69,7 +69,7 @@ describe('Subscriber', () => {
             }
         })
         const publisherIdentity = EthereumKeyPairIdentity.fromPrivateKey(publisherWallet.privateKey)
-        messageSigner = new MessageSigner(publisherIdentity)
+        messageSigner = createMessageSigner(publisherIdentity)
     })
 
     afterAll(async () => {
diff --git a/packages/sdk/test/integration/gap-fill.test.ts b/packages/sdk/test/integration/gap-fill.test.ts
@@ -4,10 +4,9 @@ import { Wallet } from 'ethers'
 import { mock } from 'jest-mock-extended'
 import { GroupKey } from '../../src/encryption/GroupKey'
 import { StreamMessage } from '../../src/protocol/StreamMessage'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
 import { FakeEnvironment } from '../test-utils/fake/FakeEnvironment'
-import { createGroupKeyQueue, createStreamRegistry, createTestStream, startFailingStorageNode } from '../test-utils/utils'
+import { createGroupKeyQueue, createMessageSigner, createStreamRegistry, createTestStream, startFailingStorageNode } from '../test-utils/utils'
 import { Stream } from './../../src/Stream'
 import { MessageFactory } from './../../src/publish/MessageFactory'
 import { EthereumKeyPairIdentity } from '../../src/identity/EthereumKeyPairIdentity'
@@ -45,7 +44,7 @@ describe('gap fill', () => {
             streamRegistry: createStreamRegistry(),
             groupKeyQueue: await createGroupKeyQueue(identity, GROUP_KEY),
             signatureValidator: mock<SignatureValidator>(),
-            messageSigner: new MessageSigner(identity),
+            messageSigner: createMessageSigner(identity),
             config: createStrictConfig()
         })
     })
diff --git a/packages/sdk/test/integration/parallel-key-exchange.test.ts b/packages/sdk/test/integration/parallel-key-exchange.test.ts
@@ -9,9 +9,8 @@ import { GroupKey } from '../../src/encryption/GroupKey'
 import { StreamPermission } from '../../src/permission'
 import { StreamMessageType } from '../../src/protocol/StreamMessage'
 import { MessageFactory } from '../../src/publish/MessageFactory'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
-import { createGroupKeyQueue, createStreamRegistry } from '../test-utils/utils'
+import { createGroupKeyQueue, createMessageSigner, createStreamRegistry } from '../test-utils/utils'
 import { FakeEnvironment } from './../test-utils/fake/FakeEnvironment'
 import { EthereumKeyPairIdentity } from '../../src/identity/EthereumKeyPairIdentity'
 import { createStrictConfig } from '../../src/Config'
@@ -73,7 +72,7 @@ describe('parallel key exchange', () => {
                 }),
                 groupKeyQueue: await createGroupKeyQueue(identity, publisher.groupKey),
                 signatureValidator: mock<SignatureValidator>(),
-                messageSigner: new MessageSigner(identity),
+                messageSigner: createMessageSigner(identity),
                 config: createStrictConfig()
             })
             for (let i = 0; i < MESSAGE_COUNT_PER_PUBLISHER; i++) {
diff --git a/packages/sdk/test/integration/waitForStorage.test.ts b/packages/sdk/test/integration/waitForStorage.test.ts
@@ -6,7 +6,7 @@ import { StreamPermission } from '../../src/permission'
 import { MessageSigner } from '../../src/signature/MessageSigner'
 import { FakeEnvironment } from '../test-utils/fake/FakeEnvironment'
 import { FakeStorageNode } from '../test-utils/fake/FakeStorageNode'
-import { MOCK_CONTENT, createRandomIdentity, createRelativeTestStreamId } from '../test-utils/utils'
+import { MOCK_CONTENT, createMessageSigner, createRandomIdentity, createRelativeTestStreamId } from '../test-utils/utils'
 import { MessageID } from './../../src/protocol/MessageID'
 import { StreamMessageType } from './../../src/protocol/StreamMessage'
 import { randomUserId } from '@streamr/test-utils'
@@ -23,7 +23,7 @@ describe('waitForStorage', () => {
     let environment: FakeEnvironment
 
     beforeEach(async () => {
-        messageSigner = new MessageSigner(await createRandomIdentity())
+        messageSigner = createMessageSigner(await createRandomIdentity())
         environment = new FakeEnvironment()
         client = environment.createClient()
         stream = await client.createStream({
diff --git a/packages/sdk/test/test-utils/utils.ts b/packages/sdk/test/test-utils/utils.ts
@@ -49,6 +49,8 @@ import { StreamMessage } from '../../src/protocol/StreamMessage'
 import { GroupKeyQueue } from '../../src/publish/GroupKeyQueue'
 import { MessageFactory } from '../../src/publish/MessageFactory'
 import { MessageSigner } from '../../src/signature/MessageSigner'
+import { SigningService } from '../../src/signature/SigningService'
+import { createSignatureFromData } from '../../src/signature/signingUtils'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
 import { LoggerFactory } from '../../src/utils/LoggerFactory'
 import { counterId } from '../../src/utils/utils'
@@ -59,6 +61,25 @@ import { StreamIDBuilder } from '../../src/StreamIDBuilder'
 
 const logger = new Logger('sdk-test-utils')
 
+/**
+ * Creates a mock SigningService that performs signing synchronously on the main thread.
+ * Use this in tests instead of the real SigningService which spawns a worker.
+ */
+export function createMockSigningService(): SigningService {
+    return {
+        sign: createSignatureFromData,
+        destroy: () => {}
+    } as unknown as SigningService
+}
+
+/**
+ * Creates a MessageSigner for testing purposes.
+ * Uses a mock SigningService that doesn't spawn a worker.
+ */
+export function createMessageSigner(identity: Identity): MessageSigner {
+    return new MessageSigner(identity, createMockSigningService())
+}
+
 export function mockLoggerFactory(clientId?: string): LoggerFactory {
     return new LoggerFactory({
         id: clientId ?? counterId('TestCtx'),
@@ -152,7 +173,7 @@ export const createMockMessage = async (
         }),
         groupKeyQueue: await createGroupKeyQueue(identity, opts.encryptionKey, opts.nextEncryptionKey),
         signatureValidator: mock<SignatureValidator>(),
-        messageSigner: new MessageSigner(identity)
+        messageSigner: createMessageSigner(identity)
     })
     const DEFAULT_CONTENT = {}
     const plainContent = opts.content ?? DEFAULT_CONTENT
diff --git a/packages/sdk/test/unit/MessageFactory.test.ts b/packages/sdk/test/unit/MessageFactory.test.ts
@@ -9,9 +9,8 @@ import { GroupKey } from '../../src/encryption/GroupKey'
 import { GroupKeyQueue } from '../../src/publish/GroupKeyQueue'
 import { MessageFactory, MessageFactoryOptions } from '../../src/publish/MessageFactory'
 import { PublishMetadata } from '../../src/publish/Publisher'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
-import { createGroupKeyQueue, createStreamRegistry } from '../test-utils/utils'
+import { createGroupKeyQueue, createMessageSigner, createStreamRegistry } from '../test-utils/utils'
 import { StreamMessage, StreamMessageType } from './../../src/protocol/StreamMessage'
 import { EthereumKeyPairIdentity } from '../../src/identity/EthereumKeyPairIdentity'
 import { EncryptionType, SignatureType, ContentType } from '@streamr/trackerless-network'
@@ -60,7 +59,7 @@ describe('MessageFactory', () => {
                     }),
                     groupKeyQueue: await createGroupKeyQueue(identity, GROUP_KEY),
                     signatureValidator: new SignatureValidator(opts?.erc1271ContractFacade ?? mock<ERC1271ContractFacade>(), new DestroySignal()),
-                    messageSigner: new MessageSigner(identity),
+                    messageSigner: createMessageSigner(identity),
                     config: {
                         validation: {
                             permissions: true,
diff --git a/packages/sdk/test/unit/MessageStream.test.ts b/packages/sdk/test/unit/MessageStream.test.ts
@@ -3,7 +3,7 @@ import omit from 'lodash/omit'
 import { MessageSigner } from '../../src/signature/MessageSigner'
 import { MessageStream } from '../../src/subscribe/MessageStream'
 import { Msg } from '../test-utils/publish'
-import { createRandomIdentity, waitForCalls } from '../test-utils/utils'
+import { createMessageSigner, createRandomIdentity, waitForCalls } from '../test-utils/utils'
 import { convertStreamMessageToMessage } from './../../src/Message'
 import { MessageID } from './../../src/protocol/MessageID'
 import { StreamMessageType } from './../../src/protocol/StreamMessage'
@@ -28,7 +28,7 @@ describe('MessageStream', () => {
     }
 
     beforeEach(async () => {
-        messageSigner = new MessageSigner(await createRandomIdentity())
+        messageSigner = createMessageSigner(await createRandomIdentity())
     })
 
     it('onMessage', async () => {
diff --git a/packages/sdk/test/unit/PushPipeline.test.ts b/packages/sdk/test/unit/PushPipeline.test.ts
@@ -6,7 +6,7 @@ import { PushPipeline } from '../../src/utils/PushPipeline'
 import { counterId, instanceId } from '../../src/utils/utils'
 import { LeaksDetector } from '../test-utils/LeaksDetector'
 import { Msg } from '../test-utils/publish'
-import { createRandomIdentity } from '../test-utils/utils'
+import { createMessageSigner, createRandomIdentity } from '../test-utils/utils'
 import { MessageID } from './../../src/protocol/MessageID'
 import { StreamMessage, StreamMessageType } from './../../src/protocol/StreamMessage'
 import { ContentType, EncryptionType, SignatureType } from '@streamr/trackerless-network'
@@ -31,7 +31,7 @@ describe('PushPipeline', () => {
 
     beforeEach(async () => {
         leaksDetector = new LeaksDetector()
-        messageSigner = new MessageSigner(await createRandomIdentity())
+        messageSigner = createMessageSigner(await createRandomIdentity())
     })
 
     afterEach(async () => {
diff --git a/packages/sdk/test/unit/messagePipeline.test.ts b/packages/sdk/test/unit/messagePipeline.test.ts
@@ -11,11 +11,10 @@ import { GroupKey } from '../../src/encryption/GroupKey'
 import { GroupKeyManager } from '../../src/encryption/GroupKeyManager'
 import { SubscriberKeyExchange } from '../../src/encryption/SubscriberKeyExchange'
 import { StreamrClientEventEmitter } from '../../src/events'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
 import { createMessagePipeline } from '../../src/subscribe/messagePipeline'
 import { PushPipeline } from '../../src/utils/PushPipeline'
-import { mockLoggerFactory } from '../test-utils/utils'
+import { createMessageSigner, mockLoggerFactory } from '../test-utils/utils'
 import { MessageID } from './../../src/protocol/MessageID'
 import { StreamMessage, StreamMessageType } from './../../src/protocol/StreamMessage'
 import { EncryptionType, ContentType, SignatureType } from '@streamr/trackerless-network'
@@ -40,7 +39,7 @@ describe('messagePipeline', () => {
         contentType?: ContentType
     } = {}): Promise<StreamMessage> => {
         const [streamId, partition] = StreamPartIDUtils.getStreamIDAndPartition(streamPartId)
-        const messageSigner = new MessageSigner(EthereumKeyPairIdentity.fromPrivateKey(publisher.privateKey))
+        const messageSigner = createMessageSigner(EthereumKeyPairIdentity.fromPrivateKey(publisher.privateKey))
         return messageSigner.createSignedMessage({
             messageId: new MessageID(
                 streamId,
diff --git a/packages/sdk/test/unit/resendSubscription.test.ts b/packages/sdk/test/unit/resendSubscription.test.ts
@@ -6,13 +6,12 @@ import isEqual from 'lodash/isEqual'
 import last from 'lodash/last'
 import { Message } from '../../src/Message'
 import { MessageFactory } from '../../src/publish/MessageFactory'
-import { MessageSigner } from '../../src/signature/MessageSigner'
 import { SignatureValidator } from '../../src/signature/SignatureValidator'
 import { ResendRangeOptions } from '../../src/subscribe/Resends'
 import { Subscription, SubscriptionEvents } from '../../src/subscribe/Subscription'
 import { initResendSubscription } from '../../src/subscribe/resendSubscription'
 import { PushPipeline } from '../../src/utils/PushPipeline'
-import { createGroupKeyQueue, createRandomIdentity, createStreamRegistry, mockLoggerFactory } from '../test-utils/utils'
+import { createGroupKeyQueue, createMessageSigner, createRandomIdentity, createStreamRegistry, mockLoggerFactory } from '../test-utils/utils'
 import { StreamMessage } from './../../src/protocol/StreamMessage'
 import { createStrictConfig } from '../../src/Config'
 
@@ -61,7 +60,7 @@ describe('resend subscription', () => {
             }),
             groupKeyQueue: await createGroupKeyQueue(identity),
             signatureValidator: mock<SignatureValidator>(),
-            messageSigner: new MessageSigner(identity),
+            messageSigner: createMessageSigner(identity),
             config: createStrictConfig(),
         })
     })
diff --git a/packages/sdk/test/unit/validateStreamMessage2.test.ts b/packages/sdk/test/unit/validateStreamMessage2.test.ts
diff --git a/packages/sdk/test/unit/waitForAssignmentsToPropagate.test.ts b/packages/sdk/test/unit/waitForAssignmentsToPropagate.test.ts
