Skip to content

coordination: fix reconnect hang on half-open TCP connection #652

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
krasnovdm wants to merge 2 commits into
base: master
Choose a base branch
from
+19 −2
Open

coordination: fix reconnect hang on half-open TCP connection #652

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1795c95
coordination: fix reconnect hang on half-open TCP connection
dmitrybitquery May 21, 2026
06968e9
coordination: add backward-compatible Stream constructor overload, re…
dmitrybitquery May 22, 2026
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
4 changes: 2 additions & 2 deletions coordination/src/main/java/tech/ydb/coordination/impl/SessionImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ public CompletableFuture<Status> stop() {
public CompletableFuture<Status> connect() {
SessionState local = state.get();
// create new stream to connect
final Stream stream = new Stream(rpc);
final Stream stream = new Stream(rpc, connectTimeout);
if (!updateState(local, makeConnectionState(local, stream))) {
logger.warn("{} cannot be connected with state {}", this, local.getState());
stream.cancelStream();
Expand Down Expand Up @@ -214,7 +214,7 @@ private void restoreSession(long disconnectedAt, int retryCount, List<StreamMsg<
return;
}

Stream stream = new Stream(rpc);
Stream stream = new Stream(rpc, connectTimeout);
if (!updateState(local, makeConnectionState(local, stream))) {
logger.warn("{} cannot be reconnected with state {}", this, state.get().getState());
completeMessagesWithBadSession(messagesToRetry);
Expand Down
17 changes: 17 additions & 0 deletions coordination/src/main/java/tech/ydb/coordination/impl/Stream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,10 @@ class Stream {
private final Map<Long, StreamMsg<?>> messages = new ConcurrentHashMap<>();

Stream(Rpc rpc) {
this(rpc, Duration.ofSeconds(5));
}

Stream(Rpc rpc, Duration connectTimeout) {
this.scheduler = rpc.getScheduler();
this.stream = rpc.createSession(GrpcRequestSettings.newBuilder()
.disableDeadline()
Expand All @@ -53,6 +57,19 @@ class Stream {
stopFuture.complete(status);
}
});

// Guard against a half-open TCP connection where the gRPC stream never delivers
// SessionStarted. Without this timeout the reconnect loop stalls indefinitely,
// keeping CompletableFutures for pending operations (e.g. acquireEphemeralSemaphore)
// unresolved even after the application-level acquire timeout has expired.
scheduler.schedule(() -> {
if (startFuture.isDone()) {
return;
}
logger.warn("stream {} connect timeout after {} ms, cancelling", hashCode(), connectTimeout.toMillis());
stream.cancel();
startFuture.complete(Result.fail(Status.of(StatusCode.TIMEOUT)));
}, connectTimeout.toMillis(), TimeUnit.MILLISECONDS);
Comment on lines +61 to +72
}

public CompletableFuture<Status> getFinishedFuture() {
Expand Down