android: direct TCP relay gets real flow control: a socket at EOF is no longer polled for input (the engine thread spun until the app took the last byte), reading pauses at 64 KB queued for the app instead of aborting the flow at 256 KB, the app's send window reopens only for what the socket accepted (large uploads over a slow uplink were reset), and acknowledged data pulls the rest at once instead of on the 2 s poll

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

src/frontends/android/app/src/main/jni/libandroidbridge/routing/direct_tcp.c

@@ -59,6 +59,13 @@
  */
 #define MAX_BUFFERED	(256 * 1024)
 
+/**
+ * Stop reading from the remote socket while this much is waiting for the
+ * app: the data stays in the kernel socket and TCP throttles the sender,
+ * instead of the flow hitting MAX_BUFFERED and being aborted
+ */
+#define RELAY_HIGH_WATER	(64 * 1024)
+
 /**
  * Give up on a direct connect after this many milliseconds (a destination
  * that silently drops SYNs would otherwise hang the app's connection
@@ -126,6 +133,7 @@ static lwip_err_t on_accept(void *arg, struct tcp_pcb *newpcb, lwip_err_t err);
 static lwip_err_t on_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, lwip_err_t err);
 static void on_err(void *arg, lwip_err_t err);
 static lwip_err_t on_poll(void *arg, struct tcp_pcb *pcb);
+static lwip_err_t on_sent(void *arg, struct tcp_pcb *pcb, u16_t len);
 
 /**
  * One relayed connection
@@ -330,13 +338,20 @@ static void relay_flush_to_sock(tcp_relay_t *relay)
 {
 	while (relay->to_sock_len)
 	{
-		ssize_t n = write(relay->fd, relay->to_sock, relay->to_sock_len);
+		ssize_t n = write(relay->fd, relay->to_sock,
+						  min(relay->to_sock_len, 0xffff));
 
 		if (n < 0)
 		{
 			return;
 		}
 		buf_consume(relay->to_sock, &relay->to_sock_len, n);
+		if (relay->pcb)
+		{	/* reopen the app's send window only for what the socket took:
+			 * acknowledging on receipt let a fast app outrun a slow uplink
+			 * until the flow hit MAX_BUFFERED and was aborted */
+			tcp_recved(relay->pcb, n);
+		}
 	}
 }
 
@@ -729,6 +744,7 @@ static lwip_err_t on_accept(void *arg, struct tcp_pcb *newpcb, lwip_err_t err)
 	tcp_arg(newpcb, relay);
 	tcp_recv(newpcb, on_recv);
 	tcp_err(newpcb, on_err);
+	tcp_sent(newpcb, on_sent);
 	tcp_poll(newpcb, on_poll, 4);
 	this->relays->insert_last(this->relays, relay);
 	DIRECT_STATS_ADD(this->stats->tcp_flows, 1);
@@ -767,7 +783,6 @@ static lwip_err_t on_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, lwip_e
 		relay->dead = TRUE;
 		return ERR_ABRT;
 	}
-	tcp_recved(pcb, p->tot_len);
 	pbuf_free(p);
 	relay_flush_to_sock(relay);
 	return ERR_OK;
@@ -802,6 +817,22 @@ static lwip_err_t on_poll(void *arg, struct tcp_pcb *pcb)
 	return ERR_OK;
 }
 
+/**
+ * lwIP sent callback: the app acknowledged data, so the send buffer has
+ * room again.  Without it the rest of a response only moved on the next
+ * socket read or the 2 s poll callback.
+ */
+static lwip_err_t on_sent(void *arg, struct tcp_pcb *pcb, u16_t len)
+{
+	tcp_relay_t *relay = arg;
+
+	if (relay && relay->pcb && !relay->dead)
+	{
+		relay_flush_to_pcb(relay);
+	}
+	return ERR_OK;
+}
+
 /**
  * netif output: a finished IP packet for the kernel.  lwIP keeps ownership
  * of the pbuf (unacked segments reference it for retransmission and free
@@ -1020,7 +1051,16 @@ METHOD(direct_tcp_t, fill_poll, size_t,
 	while (n < cap && enumerator->enumerate(enumerator, &relay))
 	{
 		fds[n].fd = relay->fd;
-		fds[n].events = POLLIN | (relay->connected ? 0 : POLLOUT);
+		fds[n].events = 0;
+		if (!relay->connected || relay->to_sock_len)
+		{	/* connect result, or room for data the socket refused before */
+			fds[n].events |= POLLOUT;
+		}
+		if (!relay->remote_eof && relay->to_pcb_len < RELAY_HIGH_WATER)
+		{	/* a socket at EOF stays readable forever: polling it for input
+			 * made the engine thread spin until the app took the last byte */
+			fds[n].events |= POLLIN;
+		}
 		fds[n].revents = 0;
 		n++;
 	}
@@ -1057,6 +1097,13 @@ METHOD(direct_tcp_t, process_poll, void,
 			relay_destroy(this, relay, TRUE);
 			continue;
 		}
+		if (relay->remote_eof && (re & (POLLERR | POLLHUP)))
+		{	/* fully closed: HUP is reported whatever we ask for, so drop the
+			 * socket (poll ignores fd -1); what is buffered still goes to the app */
+			close(relay->fd);
+			relay->fd = -1;
+			re = 0;
+		}
 		if (!relay->connected)
 		{
 			uint64_t now = now_ms();
@@ -1142,13 +1189,16 @@ METHOD(direct_tcp_t, process_poll, void,
 				}
 			}
 		}
-		if (re & (POLLIN | POLLHUP))
+		if (re & (POLLIN | POLLHUP | POLLERR))
 		{
 			uint8_t buf[16384];
-			ssize_t n;
+			ssize_t n = 1;
 			bool failed = FALSE;
 
-			while ((n = read(relay->fd, buf, sizeof(buf))) > 0)
+			/* a broken socket is read out whatever is buffered: its error
+			 * must surface, or poll would report it round after round */
+			while ((relay->to_pcb_len < RELAY_HIGH_WATER || (re & (POLLHUP | POLLERR))) &&
+				   (n = read(relay->fd, buf, sizeof(buf))) > 0)
 			{
 				if (!buf_append(relay, FALSE, buf, n))
 				{