socketTimeout bounds each socket read rather than the attempt, so a slow-feeding server can keep a command alive past totalTimeout

/*
 * Copyright 2012-2026 Aerospike, Inc.
 *
 * Portions may be licensed to Aerospike, Inc. under one or more contributor
 * license agreements WHICH ARE COMPATIBLE WITH THE APACHE LICENSE, VERSION 2.0.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.aerospike.client;

import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.aerospike.client.cluster.Connection;

/**
 * socketTimeout is applied to each socket read rather than to the attempt as a whole.
 * Connection.readFully loops on in.read(...) and only that individual call is bounded, so a peer
 * that emits a little data before each socketTimeout expiry keeps the loop running. SyncCommand
 * consults its deadline only between retries, so totalTimeout never fires for such an attempt.
 * <p>
 * This test needs no Aerospike server. Connection is a socket wrapper, so a plain ServerSocket
 * that drips bytes is enough to hold readFully open indefinitely.
 */
public class ConnectionReadDeadlineTest {
	/** Bounds each individual read. No single read ever reaches it. */
	private static final int SOCKET_TIMEOUT_MILLIS = 500;

	/** Comfortably below SOCKET_TIMEOUT_MILLIS, so every read returns a byte instead of timing out. */
	private static final long DRIP_INTERVAL_MILLIS = 250;

	/** Far more than the drip can deliver, so the read loop cannot complete on its own. */
	private static final int BYTES_REQUESTED = 10_000;

	/** How long the caller is given before the read is declared unbounded. */
	private static final long OBSERVATION_MILLIS = 3_000;

	private ServerSocket server;
	private Thread dripper;
	private volatile boolean running = true;

	@Before
	public void startDripServer() throws Exception {
		server = new ServerSocket(0, 1);

		dripper = new Thread(() -> {
			try (Socket accepted = server.accept()) {
				OutputStream out = accepted.getOutputStream();

				while (running) {
					out.write(0);
					out.flush();
					Thread.sleep(DRIP_INTERVAL_MILLIS);
				}
			}
			catch (Exception e) {
				// The test closes the socket underneath this thread when it finishes.
			}
		});
		dripper.setDaemon(true);
		dripper.start();
	}

	@After
	public void stopDripServer() throws Exception {
		running = false;

		if (dripper != null) {
			dripper.interrupt();
		}

		if (server != null) {
			server.close();
		}
	}

	@Test
	public void readIsBoundedInTotalAndNotOnlyPerRead() throws Exception {
		InetSocketAddress address = new InetSocketAddress("127.0.0.1", server.getLocalPort());
		Connection conn = new Connection(address, SOCKET_TIMEOUT_MILLIS);

		CountDownLatch finished = new CountDownLatch(1);
		AtomicReference<Throwable> thrown = new AtomicReference<>();

		Thread caller = new Thread(() -> {
			try {
				conn.readFully(new byte[BYTES_REQUESTED], BYTES_REQUESTED);
			}
			catch (Throwable t) {
				thrown.set(t);
			}
			finally {
				finished.countDown();
			}
		});
		caller.setDaemon(true);
		caller.start();

		boolean returned = finished.await(OBSERVATION_MILLIS, TimeUnit.MILLISECONDS);

		try {
			if (! returned) {
				fail("readFully was still running after " + OBSERVATION_MILLIS + "ms with socketTimeout="
					+ SOCKET_TIMEOUT_MILLIS + "ms. A byte arriving every " + DRIP_INTERVAL_MILLIS
					+ "ms resets the per-read timeout, and nothing bounds the attempt as a whole, "
					+ "so the command outlives any totalTimeout the caller configured.");
			}

			assertTrue("readFully should have reported a timeout, but it returned normally",
				thrown.get() != null);
		}
		finally {
			caller.interrupt();
			conn.close();
		}
	}
}