Signal/Signal-Server
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Server synced 2026-04-20 03:08:04 +01:00
Code Issues Packages Projects Releases Wiki Activity

Retire the legacy client presence system

Browse Source
This commit is contained in:
Jon Chambers
2024-11-05 12:34:27 -05:00
committed by Jon Chambers
parent 9898e18ae2
commit 1c167ec150
27 changed files with 40 additions and 1034 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
View File

@@ -191,7 +191,6 @@ import org.whispersystems.textsecuregcm.metrics.TrafficSource;
import org.whispersystems.textsecuregcm.providers.MultiRecipientMessageProvider;
import org.whispersystems.textsecuregcm.providers.RedisClusterHealthCheck;
import org.whispersystems.textsecuregcm.push.APNSender;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.FcmSender;
import org.whispersystems.textsecuregcm.push.MessageSender;
import org.whispersystems.textsecuregcm.push.ProvisioningManager;
@@ -543,11 +542,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
.minThreads(8)
.maxThreads(8)
.build();
ExecutorService clientPresenceExecutor = environment.lifecycle()
.executorService(name(getClass(), "clientPresence-%d"))
.minThreads(8)
.maxThreads(8)
.build();
// unbounded executor (same as cachedThreadPool)
ExecutorService remoteStorageHttpExecutor = environment.lifecycle()
.executorService(name(getClass(), "remoteStorage-%d"))
@@ -617,8 +611,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
secureValueRecoveryServiceExecutor, secureValueRecoveryServiceRetryExecutor, config.getSvr2Configuration());
SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
storageServiceExecutor, storageServiceRetryExecutor, config.getSecureStorageServiceConfiguration());
ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster, recurringJobExecutor,
keyspaceNotificationDispatchExecutor);
PubSubClientEventManager pubSubClientEventManager = new PubSubClientEventManager(messagesCluster, clientEventExecutor);
ProfilesManager profilesManager = new ProfilesManager(profiles, cacheCluster);
MessagesCache messagesCache = new MessagesCache(messagesCluster, keyspaceNotificationDispatchExecutor,
@@ -637,9 +629,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
new ClientPublicKeysManager(clientPublicKeys, accountLockManager, accountLockExecutor);
AccountsManager accountsManager = new AccountsManager(accounts, phoneNumberIdentifiers, cacheCluster,
pubsubClient, accountLockManager, keysManager, messagesManager, profilesManager,
secureStorageClient, secureValueRecovery2Client,
clientPresenceManager, pubSubClientEventManager,
registrationRecoveryPasswordsManager, clientPublicKeysManager, accountLockExecutor, clientPresenceExecutor,
secureStorageClient, secureValueRecovery2Client, pubSubClientEventManager,
registrationRecoveryPasswordsManager, clientPublicKeysManager, accountLockExecutor,
clock, config.getLinkDeviceSecretConfiguration().secret().value(), dynamicConfigurationManager);
RemoteConfigsManager remoteConfigsManager = new RemoteConfigsManager(remoteConfigs);
APNSender apnSender = new APNSender(apnSenderExecutor, config.getApnConfiguration());
@@ -668,7 +659,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
new MessageDeliveryLoopMonitor(rateLimitersCluster);
final RegistrationLockVerificationManager registrationLockVerificationManager = new RegistrationLockVerificationManager(
accountsManager, clientPresenceManager, pubSubClientEventManager, svr2CredentialsGenerator, svr3CredentialsGenerator,
accountsManager, pubSubClientEventManager, svr2CredentialsGenerator, svr3CredentialsGenerator,
registrationRecoveryPasswordsManager, pushNotificationManager, rateLimiters);
final ReportedMessageMetricsListener reportedMessageMetricsListener = new ReportedMessageMetricsListener(
@@ -745,7 +736,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
environment.lifecycle().manage(pushNotificationScheduler);
environment.lifecycle().manage(provisioningManager);
environment.lifecycle().manage(messagesCache);
environment.lifecycle().manage(clientPresenceManager);
environment.lifecycle().manage(pubSubClientEventManager);
environment.lifecycle().manage(currencyManager);
environment.lifecycle().manage(registrationServiceClient);
@@ -998,8 +988,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
environment.jersey().register(MultiRecipientMessageProvider.class);
environment.jersey().register(new AuthDynamicFeature(accountAuthFilter));
environment.jersey().register(new AuthValueFactoryProvider.Binder<>(AuthenticatedDevice.class));
environment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager, clientPresenceManager,
pubSubClientEventManager));
environment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager, pubSubClientEventManager));
environment.jersey().register(new TimestampResponseFilter());
///
@@ -1009,11 +998,10 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
webSocketEnvironment.setAuthenticator(new WebSocketAccountAuthenticator(accountAuthenticator, new AccountPrincipalSupplier(accountsManager)));
webSocketEnvironment.setConnectListener(
new AuthenticatedConnectListener(receiptSender, messagesManager, messageMetrics, pushNotificationManager,
pushNotificationScheduler, clientPresenceManager, pubSubClientEventManager, websocketScheduledExecutor,
pushNotificationScheduler, pubSubClientEventManager, websocketScheduledExecutor,
messageDeliveryScheduler, clientReleaseManager, messageDeliveryLoopMonitor));
webSocketEnvironment.jersey()
.register(new WebsocketRefreshApplicationEventListener(accountsManager, clientPresenceManager,
pubSubClientEventManager));
.register(new WebsocketRefreshApplicationEventListener(accountsManager, pubSubClientEventManager));
webSocketEnvironment.jersey().register(new RateLimitByIpFilter(rateLimiters));
webSocketEnvironment.jersey().register(new RequestStatisticsFilter(TrafficSource.WEBSOCKET));
webSocketEnvironment.jersey().register(MultiRecipientMessageProvider.class);
@@ -1155,8 +1143,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
WebSocketEnvironment<AuthenticatedDevice> provisioningEnvironment = new WebSocketEnvironment<>(environment,
webSocketEnvironment.getRequestLog(), Duration.ofMillis(60000));
provisioningEnvironment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager, clientPresenceManager,
pubSubClientEventManager));
provisioningEnvironment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager, pubSubClientEventManager));
provisioningEnvironment.setConnectListener(new ProvisioningConnectListener(provisioningManager));
provisioningEnvironment.jersey().register(new MetricsApplicationEventListener(TrafficSource.WEBSOCKET, clientReleaseManager));
provisioningEnvironment.jersey().register(new KeepAliveController(pubSubClientEventManager));

5
service/src/main/java/org/whispersystems/textsecuregcm/auth/RegistrationLockVerificationManager.java
View File

@@ -25,7 +25,6 @@ import org.whispersystems.textsecuregcm.entities.RegistrationLockFailure;
import org.whispersystems.textsecuregcm.entities.Svr3Credentials;
import org.whispersystems.textsecuregcm.limits.RateLimiters;
import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.NotPushRegisteredException;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
import org.whispersystems.textsecuregcm.push.PushNotificationManager;
@@ -55,7 +54,6 @@ public class RegistrationLockVerificationManager {
private static final String PHONE_VERIFICATION_TYPE_TAG_NAME = "phoneVerificationType";
private final AccountsManager accounts;
private final ClientPresenceManager clientPresenceManager;
private final PubSubClientEventManager pubSubClientEventManager;
private final ExternalServiceCredentialsGenerator svr2CredentialGenerator;
private final ExternalServiceCredentialsGenerator svr3CredentialGenerator;
@@ -65,7 +63,6 @@ public class RegistrationLockVerificationManager {
public RegistrationLockVerificationManager(
final AccountsManager accounts,
final ClientPresenceManager clientPresenceManager,
final PubSubClientEventManager pubSubClientEventManager,
final ExternalServiceCredentialsGenerator svr2CredentialGenerator,
final ExternalServiceCredentialsGenerator svr3CredentialGenerator,
@@ -73,7 +70,6 @@ public class RegistrationLockVerificationManager {
final PushNotificationManager pushNotificationManager,
final RateLimiters rateLimiters) {
this.accounts = accounts;
this.clientPresenceManager = clientPresenceManager;
this.pubSubClientEventManager = pubSubClientEventManager;
this.svr2CredentialGenerator = svr2CredentialGenerator;
this.svr3CredentialGenerator = svr3CredentialGenerator;
@@ -165,7 +161,6 @@ public class RegistrationLockVerificationManager {
}
final List<Byte> deviceIds = updatedAccount.getDevices().stream().map(Device::getId).toList();
clientPresenceManager.disconnectAllPresences(updatedAccount.getUuid(), deviceIds);
pubSubClientEventManager.requestDisconnection(updatedAccount.getUuid(), deviceIds);
try {

5
service/src/main/java/org/whispersystems/textsecuregcm/auth/WebsocketRefreshApplicationEventListener.java
View File

@@ -9,7 +9,6 @@ import org.glassfish.jersey.server.monitoring.ApplicationEvent;
import org.glassfish.jersey.server.monitoring.ApplicationEventListener;
import org.glassfish.jersey.server.monitoring.RequestEvent;
import org.glassfish.jersey.server.monitoring.RequestEventListener;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
import org.whispersystems.textsecuregcm.storage.AccountsManager;
@@ -21,11 +20,9 @@ public class WebsocketRefreshApplicationEventListener implements ApplicationEven
private final WebsocketRefreshRequestEventListener websocketRefreshRequestEventListener;
public WebsocketRefreshApplicationEventListener(final AccountsManager accountsManager,
final ClientPresenceManager clientPresenceManager,
final PubSubClientEventManager pubSubClientEventManager) {
this.websocketRefreshRequestEventListener = new WebsocketRefreshRequestEventListener(clientPresenceManager,
pubSubClientEventManager,
this.websocketRefreshRequestEventListener = new WebsocketRefreshRequestEventListener(pubSubClientEventManager,
new LinkedDeviceRefreshRequirementProvider(accountsManager),
new PhoneNumberChangeRefreshRequirementProvider(accountsManager));
}

5
service/src/main/java/org/whispersystems/textsecuregcm/auth/WebsocketRefreshRequestEventListener.java
View File

@@ -19,12 +19,10 @@ import org.glassfish.jersey.server.monitoring.RequestEvent.Type;
import org.glassfish.jersey.server.monitoring.RequestEventListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
public class WebsocketRefreshRequestEventListener implements RequestEventListener {
private final ClientPresenceManager clientPresenceManager;
private final PubSubClientEventManager pubSubClientEventManager;
private final WebsocketRefreshRequirementProvider[] providers;
@@ -37,11 +35,9 @@ public class WebsocketRefreshRequestEventListener implements RequestEventListene
private static final Logger logger = LoggerFactory.getLogger(WebsocketRefreshRequestEventListener.class);
public WebsocketRefreshRequestEventListener(
final ClientPresenceManager clientPresenceManager,
final PubSubClientEventManager pubSubClientEventManager,
final WebsocketRefreshRequirementProvider... providers) {
this.clientPresenceManager = clientPresenceManager;
this.pubSubClientEventManager = pubSubClientEventManager;
this.providers = providers;
}
@@ -64,7 +60,6 @@ public class WebsocketRefreshRequestEventListener implements RequestEventListene
.forEach(pair -> {
try {
displacedDevices.incrementAndGet();
clientPresenceManager.disconnectPresence(pair.first(), pair.second());
pubSubClientEventManager.requestDisconnection(pair.first(), List.of(pair.second()));
} catch (final Exception e) {
logger.error("Could not displace device presence", e);

1
service/src/main/java/org/whispersystems/textsecuregcm/controllers/KeepAliveController.java
View File

@@ -22,7 +22,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.auth.AuthenticatedDevice;
import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
import org.whispersystems.websocket.auth.ReadOnly;
import org.whispersystems.websocket.session.WebSocketSession;

390
service/src/main/java/org/whispersystems/textsecuregcm/push/ClientPresenceManager.java
View File

@@ -1,390 +0,0 @@
/*
* Copyright 2013 Signal Messenger, LLC
* SPDX-License-Identifier: AGPL-3.0-only
*/
package org.whispersystems.textsecuregcm.push;
import static com.codahale.metrics.MetricRegistry.name;
import com.google.common.annotations.VisibleForTesting;
import io.dropwizard.lifecycle.Managed;
import io.lettuce.core.LettuceFutures;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.ScriptOutputType;
import io.lettuce.core.cluster.SlotHash;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import io.lettuce.core.cluster.event.ClusterTopologyChangedEvent;
import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
import io.lettuce.core.cluster.pubsub.RedisClusterPubSubAdapter;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Timer;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.redis.ClusterLuaScript;
import org.whispersystems.textsecuregcm.redis.FaultTolerantPubSubClusterConnection;
import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient;
import org.whispersystems.textsecuregcm.storage.Device;
/**
* The client presence manager keeps track of which clients are actively connected and "present" to receive messages.
* Only one client per account/device may be present at a time; if a second client for the same account/device declares
* its presence, the previous client is displaced.
* <p/>
* The client presence manager depends on Redis keyspace notifications and requires that the Redis instance support at
* least the following notification types: {@code K$z}.
*/
public class ClientPresenceManager extends RedisClusterPubSubAdapter<String, String> implements Managed {
private final String managerId = UUID.randomUUID().toString();
private final String connectedClientSetKey = getConnectedClientSetKey(managerId);
private final FaultTolerantRedisClusterClient presenceCluster;
private final FaultTolerantPubSubClusterConnection<String, String> pubSubConnection;
private final ClusterLuaScript clearPresenceScript;
private final ClusterLuaScript renewPresenceScript;
private final ExecutorService keyspaceNotificationExecutorService;
private final ScheduledExecutorService scheduledExecutorService;
private ScheduledFuture<?> pruneMissingPeersFuture;
private final Map<String, DisplacedPresenceListener> displacementListenersByPresenceKey = new ConcurrentHashMap<>();
private final Map<String, CompletionStage<?>> pendingPresenceSetsByPresenceKey = new ConcurrentHashMap<>();
private final Timer checkPresenceTimer;
private final Timer setPresenceTimer;
private final Timer clearPresenceTimer;
private final Timer prunePeersTimer;
private final Counter pruneClientMeter;
private final Counter remoteDisplacementMeter;
private final Counter pubSubMessageMeter;
private final Counter displacementListenerAlreadyRemovedCounter;
private static final int PRUNE_PEERS_INTERVAL_SECONDS = (int) Duration.ofSeconds(30).toSeconds();
private static final int PRESENCE_EXPIRATION_SECONDS = (int) Duration.ofMinutes(11).toSeconds();
static final String MANAGER_SET_KEY = "presence::managers";
private static final Logger log = LoggerFactory.getLogger(ClientPresenceManager.class);
public ClientPresenceManager(final FaultTolerantRedisClusterClient presenceCluster,
final ScheduledExecutorService scheduledExecutorService,
final ExecutorService keyspaceNotificationExecutorService) throws IOException {
this.presenceCluster = presenceCluster;
this.pubSubConnection = this.presenceCluster.createPubSubConnection();
this.clearPresenceScript = ClusterLuaScript.fromResource(presenceCluster, "lua/clear_presence.lua",
ScriptOutputType.INTEGER);
this.renewPresenceScript = ClusterLuaScript.fromResource(presenceCluster, "lua/renew_presence.lua",
ScriptOutputType.VALUE);
this.scheduledExecutorService = scheduledExecutorService;
this.keyspaceNotificationExecutorService = keyspaceNotificationExecutorService;
Metrics.gauge(name(getClass(), "localClientCount"), this, ignored -> displacementListenersByPresenceKey.size());
this.checkPresenceTimer = Metrics.timer(name(getClass(), "checkPresence"));
this.setPresenceTimer = Metrics.timer(name(getClass(), "setPresence"));
this.clearPresenceTimer = Metrics.timer(name(getClass(), "clearPresence"));
this.prunePeersTimer = Metrics.timer(name(getClass(), "prunePeers"));
this.pruneClientMeter = Metrics.counter(name(getClass(), "pruneClient"));
this.remoteDisplacementMeter = Metrics.counter(name(getClass(), "remoteDisplacement"));
this.pubSubMessageMeter = Metrics.counter(name(getClass(), "pubSubMessage"));
this.displacementListenerAlreadyRemovedCounter = Metrics.counter(
name(getClass(), "displacementListenerAlreadyRemoved"));
}
@VisibleForTesting
FaultTolerantPubSubClusterConnection<String, String> getPubSubConnection() {
return pubSubConnection;
}
@Override
public void start() {
pubSubConnection.usePubSubConnection(connection -> {
connection.addListener(this);
final String presenceChannel = getManagerPresenceChannel(managerId);
final int slot = SlotHash.getSlot(presenceChannel);
connection.sync().nodes(node -> node.is(RedisClusterNode.NodeFlag.UPSTREAM) && node.hasSlot(slot))
.commands()
.subscribe(presenceChannel);
});
pubSubConnection.subscribeToClusterTopologyChangedEvents(this::resubscribeAll);
presenceCluster.useCluster(connection -> connection.sync().sadd(MANAGER_SET_KEY, managerId));
pruneMissingPeersFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> {
try {
pruneMissingPeers();
} catch (final Throwable t) {
log.warn("Failed to prune missing peers", t);
}
}, new Random().nextInt(PRUNE_PEERS_INTERVAL_SECONDS), PRUNE_PEERS_INTERVAL_SECONDS, TimeUnit.SECONDS);
}
@Override
public void stop() {
pubSubConnection.usePubSubConnection(connection -> connection.removeListener(this));
if (pruneMissingPeersFuture != null) {
pruneMissingPeersFuture.cancel(false);
}
for (final String presenceKey : displacementListenersByPresenceKey.keySet()) {
clearPresence(presenceKey);
}
presenceCluster.useCluster(connection -> {
connection.sync().srem(MANAGER_SET_KEY, managerId);
connection.sync().del(getConnectedClientSetKey(managerId));
});
pubSubConnection.usePubSubConnection(
connection -> connection.sync().upstream().commands().unsubscribe(getManagerPresenceChannel(managerId)));
}
public void setPresent(final UUID accountUuid, final byte deviceId,
final DisplacedPresenceListener displacementListener) {
setPresenceTimer.record(() -> {
final String presenceKey = getPresenceKey(accountUuid, deviceId);
displacePresence(presenceKey, true);
displacementListenersByPresenceKey.put(presenceKey, displacementListener);
final CompletableFuture<Void> presenceFuture = new CompletableFuture<>();
final CompletionStage<?> previousFuture = pendingPresenceSetsByPresenceKey.put(presenceKey, presenceFuture);
if (previousFuture != null) {
log.debug("Another presence is already pending for {}:{}", accountUuid, deviceId);
}
subscribeForRemotePresenceChanges(presenceKey);
presenceCluster.withCluster(connection -> {
final RedisAdvancedClusterAsyncCommands<String, String> commands = connection.async();
commands.sadd(connectedClientSetKey, presenceKey);
return commands.setex(presenceKey, PRESENCE_EXPIRATION_SECONDS, managerId);
}).whenComplete((result, throwable) -> {
if (throwable != null) {
presenceFuture.completeExceptionally(throwable);
} else {
presenceFuture.complete(null);
}
});
presenceFuture.whenComplete(
(ignored, throwable) -> pendingPresenceSetsByPresenceKey.remove(presenceKey, presenceFuture));
});
}
public void renewPresence(final UUID accountUuid, final byte deviceId) {
renewPresenceScript.execute(List.of(getPresenceKey(accountUuid, deviceId)),
List.of(managerId, String.valueOf(PRESENCE_EXPIRATION_SECONDS)));
}
public void disconnectAllPresences(final UUID accountUuid, final List<Byte> deviceIds) {
List<String> presenceKeys = new ArrayList<>();
deviceIds.forEach(deviceId -> {
String presenceKey = getPresenceKey(accountUuid, deviceId);
if (isLocallyPresent(accountUuid, deviceId)) {
displacePresence(presenceKey, false);
}
presenceKeys.add(presenceKey);
});
presenceCluster.useCluster(connection -> {
List<RedisFuture<Long>> futures = presenceKeys.stream().map(key -> connection.async().del(key)).toList();
LettuceFutures.awaitAll(connection.getTimeout(), futures.toArray(new RedisFuture[0]));
});
}
public void disconnectAllPresencesForUuid(final UUID accountUuid) {
disconnectAllPresences(accountUuid, Device.ALL_POSSIBLE_DEVICE_IDS);
}
public void disconnectPresence(final UUID accountUuid, final byte deviceId) {
disconnectAllPresences(accountUuid, List.of(deviceId));
}
private void displacePresence(final String presenceKey, final boolean connectedElsewhere) {
final DisplacedPresenceListener displacementListener = displacementListenersByPresenceKey.get(presenceKey);
if (displacementListener != null) {
displacementListener.handleDisplacement(connectedElsewhere);
}
clearPresence(presenceKey);
}
public boolean isPresent(final UUID accountUuid, final byte deviceId) {
return checkPresenceTimer.record(() ->
presenceCluster.withCluster(connection ->
connection.sync().exists(getPresenceKey(accountUuid, deviceId))) == 1);
}
public boolean isLocallyPresent(final UUID accountUuid, final byte deviceId) {
return displacementListenersByPresenceKey.containsKey(getPresenceKey(accountUuid, deviceId));
}
public boolean clearPresence(final UUID accountUuid, final byte deviceId, final DisplacedPresenceListener listener) {
final String presenceKey = getPresenceKey(accountUuid, deviceId);
if (displacementListenersByPresenceKey.remove(presenceKey, listener)) {
return clearPresence(presenceKey);
} else {
displacementListenerAlreadyRemovedCounter.increment();
return false;
}
}
private boolean clearPresence(final String presenceKey) {
return clearPresenceTimer.record(() -> {
displacementListenersByPresenceKey.remove(presenceKey);
unsubscribeFromRemotePresenceChanges(presenceKey);
final boolean removed = clearPresenceScript.execute(List.of(presenceKey), List.of(managerId)) != null;
presenceCluster.useCluster(connection -> connection.sync().srem(connectedClientSetKey, presenceKey));
return removed;
});
}
private void subscribeForRemotePresenceChanges(final String presenceKey) {
final int slot = SlotHash.getSlot(presenceKey);
pubSubConnection.usePubSubConnection(
connection -> connection.sync().nodes(node -> node.is(RedisClusterNode.NodeFlag.UPSTREAM) && node.hasSlot(slot))
.commands()
.subscribe(getKeyspaceNotificationChannel(presenceKey)));
}
private void resubscribeAll(final ClusterTopologyChangedEvent event) {
for (final String presenceKey : displacementListenersByPresenceKey.keySet()) {
subscribeForRemotePresenceChanges(presenceKey);
}
}
private void unsubscribeFromRemotePresenceChanges(final String presenceKey) {
pubSubConnection.usePubSubConnection(
connection -> connection.sync().upstream().commands().unsubscribe(getKeyspaceNotificationChannel(presenceKey)));
}
void pruneMissingPeers() {
prunePeersTimer.record(() -> {
final Set<String> peerIds = presenceCluster.withCluster(
connection -> connection.sync().smembers(MANAGER_SET_KEY));
peerIds.remove(managerId);
for (final String peerId : peerIds) {
final boolean peerMissing = presenceCluster.withCluster(
connection -> connection.sync().publish(getManagerPresenceChannel(peerId), "ping") == 0);
if (peerMissing) {
log.debug("Presence manager {} did not respond to ping", peerId);
final String connectedClientsKey = getConnectedClientSetKey(peerId);
String presenceKey;
while ((presenceKey = presenceCluster.withCluster(connection -> connection.sync().spop(connectedClientsKey)))
!= null) {
clearPresenceScript.execute(List.of(presenceKey), List.of(peerId));
pruneClientMeter.increment();
}
presenceCluster.useCluster(connection -> {
connection.sync().del(connectedClientsKey);
connection.sync().srem(MANAGER_SET_KEY, peerId);
});
}
}
});
}
@Override
public void message(final RedisClusterNode node, final String channel, final String message) {
pubSubMessageMeter.increment();
if (channel.startsWith("__keyspace@0__:presence::{")) {
if ("set".equals(message) || "del".equals(message)) {
// "set" might mean the client has connected to another host, although it might just be our own `set`,
// because we subscribe for changes before setting the key.
// for "del", another process has indicated the client should be disconnected
final boolean maybeConnectedElsewhere = "set".equals(message);
// At this point, we're on a Lettuce IO thread and need to dispatch to a separate thread before making
// synchronous Lettuce calls to avoid deadlocking.
keyspaceNotificationExecutorService.execute(() -> {
final String clientPresenceKey = channel.substring("__keyspace@0__:".length());
final CompletionStage<?> pendingConnection = pendingPresenceSetsByPresenceKey.getOrDefault(clientPresenceKey,
CompletableFuture.completedFuture(null));
pendingConnection.thenCompose(ignored -> {
if (maybeConnectedElsewhere) {
return presenceCluster.withCluster(connection -> connection.async().get(clientPresenceKey))
.thenApply(currentManagerId -> !managerId.equals(currentManagerId));
}
return CompletableFuture.completedFuture(true);
})
.exceptionally(ignored -> true)
.thenAcceptAsync(shouldDisplace -> {
if (shouldDisplace) {
try {
displacePresence(clientPresenceKey, maybeConnectedElsewhere);
remoteDisplacementMeter.increment();
} catch (final Exception e) {
log.warn("Error displacing presence", e);
}
}
}, keyspaceNotificationExecutorService);
});
}
}
}
@VisibleForTesting
String getManagerId() {
return managerId;
}
@VisibleForTesting
static String getPresenceKey(final UUID accountUuid, final byte deviceId) {
return "presence::{" + accountUuid.toString() + "::" + deviceId + "}";
}
private static String getKeyspaceNotificationChannel(final String presenceKey) {
return "__keyspace@0__:" + presenceKey;
}
@VisibleForTesting
static String getConnectedClientSetKey(final String managerId) {
return "presence::clients::" + managerId;
}
@VisibleForTesting
static String getManagerPresenceChannel(final String managerId) {
return "presence::manager::" + managerId;
}
}

16
service/src/main/java/org/whispersystems/textsecuregcm/push/DisplacedPresenceListener.java
View File

@@ -1,16 +0,0 @@
/*
* Copyright 2013-2020 Signal Messenger, LLC
* SPDX-License-Identifier: AGPL-3.0-only
*/
package org.whispersystems.textsecuregcm.push;
/**
* A displaced presence listener is notified when a specific client's presence has been displaced because the same
* client opened a newer connection to the Signal service.
*/
@FunctionalInterface
public interface DisplacedPresenceListener {
void handleDisplacement(boolean connectedElsewhere);
}

28
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
View File

@@ -68,19 +68,17 @@ import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfigurati
import org.whispersystems.textsecuregcm.controllers.MismatchedDevicesException;
import org.whispersystems.textsecuregcm.entities.AccountAttributes;
import org.whispersystems.textsecuregcm.entities.DeviceInfo;
import org.whispersystems.textsecuregcm.entities.RestoreAccountRequest;
import org.whispersystems.textsecuregcm.entities.ECSignedPreKey;
import org.whispersystems.textsecuregcm.entities.KEMSignedPreKey;
import org.whispersystems.textsecuregcm.entities.RemoteAttachment;
import org.whispersystems.textsecuregcm.entities.RestoreAccountRequest;
import org.whispersystems.textsecuregcm.identity.IdentityType;
import org.whispersystems.textsecuregcm.identity.ServiceIdentifier;
import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
import org.whispersystems.textsecuregcm.redis.FaultTolerantPubSubConnection;
import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient;
import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClient;
import org.whispersystems.textsecuregcm.redis.RedisOperation;
import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecoveryException;
@@ -126,12 +124,10 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
private final ProfilesManager profilesManager;
private final SecureStorageClient secureStorageClient;
private final SecureValueRecovery2Client secureValueRecovery2Client;
private final ClientPresenceManager clientPresenceManager;
private final PubSubClientEventManager pubSubClientEventManager;
private final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager;
private final ClientPublicKeysManager clientPublicKeysManager;
private final Executor accountLockExecutor;
private final Executor clientPresenceExecutor;
private final Clock clock;
private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
@@ -206,12 +202,10 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
final ProfilesManager profilesManager,
final SecureStorageClient secureStorageClient,
final SecureValueRecovery2Client secureValueRecovery2Client,
final ClientPresenceManager clientPresenceManager,
final PubSubClientEventManager pubSubClientEventManager,
final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager,
final ClientPublicKeysManager clientPublicKeysManager,
final Executor accountLockExecutor,
final Executor clientPresenceExecutor,
final Clock clock,
final byte[] linkDeviceSecret,
final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager) {
@@ -225,12 +219,10 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
this.profilesManager = profilesManager;
this.secureStorageClient = secureStorageClient;
this.secureValueRecovery2Client = secureValueRecovery2Client;
this.clientPresenceManager = clientPresenceManager;
this.pubSubClientEventManager = pubSubClientEventManager;
this.registrationRecoveryPasswordsManager = requireNonNull(registrationRecoveryPasswordsManager);
this.clientPublicKeysManager = clientPublicKeysManager;
this.accountLockExecutor = accountLockExecutor;
this.clientPresenceExecutor = clientPresenceExecutor;
this.clock = requireNonNull(clock);
this.dynamicConfigurationManager = dynamicConfigurationManager;
@@ -333,10 +325,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
keysManager.deleteSingleUsePreKeys(pni),
messagesManager.clear(aci),
profilesManager.deleteAll(aci))
.thenRunAsync(() -> {
clientPresenceManager.disconnectAllPresencesForUuid(aci);
pubSubClientEventManager.requestDisconnection(aci);
}, clientPresenceExecutor)
.thenCompose(ignored -> pubSubClientEventManager.requestDisconnection(aci))
.thenCompose(ignored -> accounts.reclaimAccount(e.getExistingAccount(), account, additionalWriteItems))
.thenCompose(ignored -> {
// We should have cleared all messages before overwriting the old account, but more may have arrived
@@ -598,12 +587,11 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
return CompletableFuture.failedFuture(throwable);
})
.whenCompleteAsync((ignored, throwable) -> {
.whenComplete((ignored, throwable) -> {
if (throwable == null) {
RedisOperation.unchecked(() -> clientPresenceManager.disconnectPresence(accountIdentifier, deviceId));
pubSubClientEventManager.requestDisconnection(accountIdentifier, List.of(deviceId));
}
}, clientPresenceExecutor);
});
}
public Account changeNumber(final Account account,
@@ -1248,11 +1236,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
registrationRecoveryPasswordsManager.removeForNumber(account.getNumber()))
.thenCompose(ignored -> accounts.delete(account.getUuid(), additionalWriteItems))
.thenCompose(ignored -> redisDeleteAsync(account))
.thenRunAsync(() -> {
RedisOperation.unchecked(() -> clientPresenceManager.disconnectAllPresencesForUuid(account.getUuid()));
pubSubClientEventManager.requestDisconnection(account.getUuid());
}, clientPresenceExecutor);
.thenRun(() -> pubSubClientEventManager.requestDisconnection(account.getUuid()));
}
private String getAccountMapKey(String key) {

26
service/src/main/java/org/whispersystems/textsecuregcm/websocket/AuthenticatedConnectListener.java
View File

@@ -9,16 +9,12 @@ import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
import io.micrometer.core.instrument.Tags;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.auth.AuthenticatedDevice;
import org.whispersystems.textsecuregcm.limits.MessageDeliveryLoopMonitor;
import org.whispersystems.textsecuregcm.metrics.MessageMetrics;
import org.whispersystems.textsecuregcm.metrics.OpenWebSocketCounter;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
import org.whispersystems.textsecuregcm.push.PushNotificationManager;
import org.whispersystems.textsecuregcm.push.PushNotificationScheduler;
@@ -38,8 +34,6 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
private static final String AUTHENTICATED_TAG_NAME = "authenticated";
private static final long RENEW_PRESENCE_INTERVAL_MINUTES = 5;
private static final Logger log = LoggerFactory.getLogger(AuthenticatedConnectListener.class);
private final ReceiptSender receiptSender;
@@ -47,7 +41,6 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
private final MessageMetrics messageMetrics;
private final PushNotificationManager pushNotificationManager;
private final PushNotificationScheduler pushNotificationScheduler;
private final ClientPresenceManager clientPresenceManager;
private final PubSubClientEventManager pubSubClientEventManager;
private final ScheduledExecutorService scheduledExecutorService;
private final Scheduler messageDeliveryScheduler;
@@ -62,7 +55,6 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
MessageMetrics messageMetrics,
PushNotificationManager pushNotificationManager,
PushNotificationScheduler pushNotificationScheduler,
ClientPresenceManager clientPresenceManager,
PubSubClientEventManager pubSubClientEventManager,
ScheduledExecutorService scheduledExecutorService,
Scheduler messageDeliveryScheduler,
@@ -73,7 +65,6 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
this.messageMetrics = messageMetrics;
this.pushNotificationManager = pushNotificationManager;
this.pushNotificationScheduler = pushNotificationScheduler;
this.clientPresenceManager = clientPresenceManager;
this.pubSubClientEventManager = pubSubClientEventManager;
this.scheduledExecutorService = scheduledExecutorService;
this.messageDeliveryScheduler = messageDeliveryScheduler;
@@ -110,21 +101,11 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
clientReleaseManager,
messageDeliveryLoopMonitor);
final AtomicReference<ScheduledFuture<?>> renewPresenceFutureReference = new AtomicReference<>();
context.addWebsocketClosedListener((closingContext, statusCode, reason) -> {
final ScheduledFuture<?> renewPresenceFuture = renewPresenceFutureReference.get();
if (renewPresenceFuture != null) {
renewPresenceFuture.cancel(false);
}
// We begin the shutdown process by removing this client's "presence," which means it will again begin to
// receive push notifications for inbound messages. We should do this first because, at this point, the
// connection has already closed and attempts to actually deliver a message via the connection will not succeed.
// It's preferable to start sending push notifications as soon as possible.
RedisOperation.unchecked(() -> clientPresenceManager.clearPresence(auth.getAccount().getUuid(), auth.getAuthenticatedDevice().getId(), connection));
pubSubClientEventManager.handleClientDisconnected(auth.getAccount().getUuid(),
auth.getAuthenticatedDevice().getId());
@@ -153,14 +134,7 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
// Finally, we register this client's presence, which suppresses push notifications. We do this last because
// receiving extra push notifications is generally preferable to missing out on a push notification.
clientPresenceManager.setPresent(auth.getAccount().getUuid(), auth.getAuthenticatedDevice().getId(), connection);
pubSubClientEventManager.handleClientConnected(auth.getAccount().getUuid(), auth.getAuthenticatedDevice().getId(), connection);
renewPresenceFutureReference.set(scheduledExecutorService.scheduleAtFixedRate(() -> RedisOperation.unchecked(() ->
clientPresenceManager.renewPresence(auth.getAccount().getUuid(), auth.getAuthenticatedDevice().getId())),
RENEW_PRESENCE_INTERVAL_MINUTES,
RENEW_PRESENCE_INTERVAL_MINUTES,
TimeUnit.MINUTES));
} catch (final Exception e) {
log.warn("Failed to initialize websocket", e);
context.getClient().close(1011, "Unexpected error initializing connection");

18
service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java
View File

@@ -46,7 +46,6 @@ import org.whispersystems.textsecuregcm.metrics.MessageMetrics;
import org.whispersystems.textsecuregcm.metrics.MetricsUtil;
import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
import org.whispersystems.textsecuregcm.push.ClientEventListener;
import org.whispersystems.textsecuregcm.push.DisplacedPresenceListener;
import org.whispersystems.textsecuregcm.push.PushNotificationManager;
import org.whispersystems.textsecuregcm.push.PushNotificationScheduler;
import org.whispersystems.textsecuregcm.push.ReceiptSender;
@@ -64,7 +63,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
public class WebSocketConnection implements MessageAvailabilityListener, DisplacedPresenceListener, ClientEventListener {
public class WebSocketConnection implements MessageAvailabilityListener, ClientEventListener {
private static final DistributionSummary messageTime = Metrics.summary(
name(MessageController.class, "messageDeliveryDuration"));
@@ -513,24 +512,11 @@ public class WebSocketConnection implements MessageAvailabilityListener, Displac
.increment();
}
@Override
public void handleDisplacement(final boolean connectedElsewhere) {
final Tags tags = Tags.of(
UserAgentTagUtil.getPlatformTag(client.getUserAgent()),
Tag.of("connectedElsewhere", String.valueOf(connectedElsewhere)),
Tag.of(PRESENCE_MANAGER_TAG, "legacy")
);
Metrics.counter(DISPLACEMENT_COUNTER_NAME, tags).increment();
}
@Override
public void handleConnectionDisplaced(final boolean connectedElsewhere) {
final Tags tags = Tags.of(
UserAgentTagUtil.getPlatformTag(client.getUserAgent()),
Tag.of("connectedElsewhere", String.valueOf(connectedElsewhere)),
Tag.of(PRESENCE_MANAGER_TAG, "pubsub")
);
Tag.of("connectedElsewhere", String.valueOf(connectedElsewhere)));
Metrics.counter(DISPLACEMENT_COUNTER_NAME, tags).increment();

14
service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java
View File

@@ -35,7 +35,6 @@ import org.whispersystems.textsecuregcm.experiment.PushNotificationExperimentSam
import org.whispersystems.textsecuregcm.limits.RateLimiters;
import org.whispersystems.textsecuregcm.metrics.MicrometerAwsSdkMetricPublisher;
import org.whispersystems.textsecuregcm.push.APNSender;
import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.push.FcmSender;
import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
import org.whispersystems.textsecuregcm.push.PushNotificationManager;
@@ -77,7 +76,6 @@ record CommandDependencies(
ReportMessageManager reportMessageManager,
MessagesCache messagesCache,
MessagesManager messagesManager,
ClientPresenceManager clientPresenceManager,
KeysManager keysManager,
APNSender apnSender,
FcmSender fcmSender,
@@ -118,8 +116,6 @@ record CommandDependencies(
FaultTolerantRedisClient pubsubClient =
configuration.getRedisPubSubConfiguration().build("pubsub", redisClientResourcesBuilder.build());
ScheduledExecutorService recurringJobExecutor = environment.lifecycle()
.scheduledExecutorService(name(name, "recurringJob-%d")).threads(2).build();
Scheduler messageDeliveryScheduler = Schedulers.fromExecutorService(
environment.lifecycle().executorService("messageDelivery").minThreads(4).maxThreads(4).build());
ExecutorService keyspaceNotificationDispatchExecutor = environment.lifecycle()
@@ -132,8 +128,6 @@ record CommandDependencies(
.executorService(name(name, "storageService-%d")).maxThreads(8).minThreads(8).build();
ExecutorService accountLockExecutor = environment.lifecycle()
.executorService(name(name, "accountLock-%d")).minThreads(8).maxThreads(8).build();
ExecutorService clientPresenceExecutor = environment.lifecycle()
.executorService(name(name, "clientPresence-%d")).minThreads(8).maxThreads(8).build();
ExecutorService remoteStorageHttpExecutor = environment.lifecycle()
.executorService(name(name, "remoteStorage-%d"))
.minThreads(0).maxThreads(Integer.MAX_VALUE).workQueue(new SynchronousQueue<>())
@@ -215,8 +209,6 @@ record CommandDependencies(
configuration.getSvr2Configuration());
SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
storageServiceExecutor, storageServiceRetryExecutor, configuration.getSecureStorageServiceConfiguration());
ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster,
recurringJobExecutor, keyspaceNotificationDispatchExecutor);
PubSubClientEventManager pubSubClientEventManager = new PubSubClientEventManager(messagesCluster, clientEventExecutor);
MessagesCache messagesCache = new MessagesCache(messagesCluster, keyspaceNotificationDispatchExecutor,
messageDeliveryScheduler, messageDeletionExecutor, Clock.systemUTC(), dynamicConfigurationManager);
@@ -234,8 +226,8 @@ record CommandDependencies(
new ClientPublicKeysManager(clientPublicKeys, accountLockManager, accountLockExecutor);
AccountsManager accountsManager = new AccountsManager(accounts, phoneNumberIdentifiers, cacheCluster,
pubsubClient, accountLockManager, keys, messagesManager, profilesManager,
secureStorageClient, secureValueRecovery2Client, clientPresenceManager, pubSubClientEventManager,
registrationRecoveryPasswordsManager, clientPublicKeysManager, accountLockExecutor, clientPresenceExecutor,
secureStorageClient, secureValueRecovery2Client, pubSubClientEventManager,
registrationRecoveryPasswordsManager, clientPublicKeysManager, accountLockExecutor,
clock, configuration.getLinkDeviceSecretConfiguration().secret().value(), dynamicConfigurationManager);
RateLimiters rateLimiters = RateLimiters.createAndValidate(configuration.getLimitsConfiguration(),
dynamicConfigurationManager, rateLimitersCluster);
@@ -272,7 +264,6 @@ record CommandDependencies(
environment.lifecycle().manage(apnSender);
environment.lifecycle().manage(messagesCache);
environment.lifecycle().manage(clientPresenceManager);
environment.lifecycle().manage(pubSubClientEventManager);
environment.lifecycle().manage(new ManagedAwsCrt());
@@ -282,7 +273,6 @@ record CommandDependencies(
reportMessageManager,
messagesCache,
messagesManager,
clientPresenceManager,
keys,
apnSender,
fcmSender,