Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changeset/slimy-clocks-argue.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
---
"@rocket.chat/meteor": patch
"@rocket.chat/core-services": patch
---

`stopped` lifecycle method was unexpectedly synchronous when using microservices, causing our code to create race conditions.
4 changes: 2 additions & 2 deletions apps/meteor/app/search/server/search.internalService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,10 @@ class Search extends ServiceClassInternal {

const service = new Search();

settings.watch('Search.Provider', () => {
settings.watch('Search.Provider', async () => {
if (searchProviderService.activeProvider?.on) {
api.registerService(service);
} else {
api.destroyService(service);
await api.destroyService(service);
}
});
4 changes: 2 additions & 2 deletions apps/meteor/ee/server/NetworkBroker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -71,12 +71,12 @@ export class NetworkBroker implements IBroker {
return this.broker.call(method, data);
}

destroyService(instance: IServiceClass): void {
async destroyService(instance: IServiceClass): Promise<void> {
const name = instance.getName();
if (!name) {
return;
}
void this.broker.destroyService(name);
await this.broker.destroyService(name);
instance.removeAllListeners();
}

Expand Down
2 changes: 1 addition & 1 deletion apps/meteor/ee/server/startup/services.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ if (!License.hasValidLicense()) {
void License.onLicense('federation', async () => {
const federationServiceEE = await FederationServiceEE.createFederationService();
if (federationService) {
api.destroyService(federationService);
await api.destroyService(federationService);
}
api.registerService(federationServiceEE);
});
39 changes: 39 additions & 0 deletions apps/meteor/ee/tests/unit/server/NetworkBroker.tests.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
import { ServiceClass } from '@rocket.chat/core-services';
import { expect } from 'chai';
import sinon from 'sinon';

import { BrokerMocked } from '../../../../tests/mocks/server/BrokerMocked';
import { NetworkBroker } from '../../../server/NetworkBroker';

class DelayedStopBroker extends BrokerMocked {
async destroyService(name: string) {
const instance = this.services.get(name);

await new Promise((resolve) => setTimeout(resolve, 1000));

await instance.stopped();

super.destroyService(name);
}
}

const broker = new NetworkBroker(new DelayedStopBroker() as any);

describe('NetworkBroker', () => {
it('should wait services to be fully destroyed', async () => {
const stoppedStub = sinon.stub();

const instance = new (class extends ServiceClass {
name = 'test';

async stopped() {
stoppedStub();
}
})();

broker.createService(instance);
await broker.destroyService(instance);

expect(stoppedStub.called).to.be.true;
});
});
10 changes: 6 additions & 4 deletions apps/meteor/tests/mocks/server/BrokerMocked.ts
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
export class BrokerMocked {
actions: Record<string, (...params: unknown[]) => Promise<unknown>> = {};

destroyService(): void {
// no op
services: Map<string, any> = new Map();

destroyService(name: string): void {
this.services.delete(name);
}

createService(): void {
// no op
createService(instance: any): void {
this.services.set(instance.name, instance);
}

async call(method: string, data: any): Promise<any> {
Expand Down
4 changes: 2 additions & 2 deletions packages/core-services/src/LocalBroker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ export class LocalBroker implements IBroker {
return this.call(method, data);
}

destroyService(instance: ServiceClass): void {
async destroyService(instance: ServiceClass): Promise<void> {
const namespace = instance.getName();

instance.getEvents().forEach((event) => event.listeners.forEach((listener) => this.events.removeListener(event.eventName, listener)));
Expand All @@ -51,7 +51,7 @@ export class LocalBroker implements IBroker {
this.methods.delete(`${namespace}.${method}`);
}
instance.removeAllListeners();
instance.stopped();
await instance.stopped();
}

createService(instance: IServiceClass): void {
Expand Down
4 changes: 2 additions & 2 deletions packages/core-services/src/lib/Api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,13 @@ export class Api implements IApiService {
this.services.forEach((service) => this.broker?.createService(service));
}

destroyService(instance: IServiceClass): void {
async destroyService(instance: IServiceClass): Promise<void> {
if (!this.services.has(instance)) {
return;
}

if (this.broker) {
this.broker.destroyService(instance);
await this.broker.destroyService(instance);
}

this.services.delete(instance);
Expand Down
2 changes: 1 addition & 1 deletion packages/core-services/src/types/IApiService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ import type { IServiceClass } from './ServiceClass';
export interface IApiService {
setBroker(broker: IBroker): void;

destroyService(instance: IServiceClass): void;
destroyService(instance: IServiceClass): Promise<void>;

registerService(instance: IServiceClass): void;

Expand Down
2 changes: 1 addition & 1 deletion packages/core-services/src/types/IBroker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ export interface IServiceMetrics {

export interface IBroker {
metrics?: IServiceMetrics;
destroyService(service: IServiceClass): void;
destroyService(service: IServiceClass): Promise<void>;
createService(service: IServiceClass, serviceDependencies?: string[]): void;
call(method: string, data: any): Promise<any>;
waitAndCall(method: string, data: any): Promise<any>;
Expand Down