Recover VPN fabric lease expiry

agents/rap-node-agent/internal/agent/payload.go

@@ -7,7 +7,7 @@ import (
 	"github.com/example/remote-access-platform/agents/rap-node-agent/internal/state"
 )
 
-const Version = "0.2.277-vpnpribatch20"
+const Version = "0.2.278-vpnpreempt"
 
 func EnrollmentPayload(clusterID, joinToken string, identity state.Identity) client.EnrollRequest {
 	return client.EnrollRequest{

agents/rap-node-agent/internal/vpnruntime/gateway.go

@@ -350,8 +350,13 @@ func (g *Gateway) uploadGatewayPackets(ctx context.Context, priorityPackets <-ch
 		return true
 	}
 	flushPriority := func(packet []byte) {
-		flush()
+		pendingBatch := batch
+		pendingBatchBytes := batchBytes
+		batch = make([][]byte, 0, vpnGatewayBatchMaxPackets)
+		batchBytes = 0
 		if !addPacket(packet) {
+			batch = pendingBatch
+			batchBytes = pendingBatchBytes
 			return
 		}
 		deadline := time.Now().Add(vpnGatewayPriorityBatchWait)
@@ -379,6 +384,14 @@ func (g *Gateway) uploadGatewayPackets(ctx context.Context, priorityPackets <-ch
 			}
 		}
 		flush()
+		if len(pendingBatch) > 0 {
+			batch = pendingBatch
+			batchBytes = pendingBatchBytes
+			if !timerActive {
+				timer.Reset(vpnGatewayBatchFlushTimeout)
+				timerActive = true
+			}
+		}
 	}
 	for {
 		if len(batch) == 0 && timerActive {

agents/rap-node-agent/internal/vpnruntime/gateway_test.go

@@ -76,6 +76,47 @@ func TestGatewayUploadPrioritizesTCPControlPackets(t *testing.T) {
 	}
 }
 
+func TestGatewayUploadPreemptsPendingNormalBatchForTCPControlPackets(t *testing.T) {
+	transport := &recordingGatewayTransport{}
+	gateway := &Gateway{Transport: transport, VPNConnectionID: "vpn-1"}
+	priorityPackets := make(chan []byte, 1)
+	packets := make(chan []byte, 1)
+
+	normal := testIPv4TCPPacket([4]byte{101, 32, 118, 25}, [4]byte{10, 77, 0, 2}, 443, 37566)
+	priority := testIPv4TCPPacket([4]byte{192, 168, 200, 95}, [4]byte{10, 77, 0, 2}, 3389, 51000)
+	priority[33] = 0x12
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- gateway.uploadGatewayPackets(ctx, priorityPackets, packets)
+	}()
+	defer func() {
+		cancel()
+		<-done
+	}()
+
+	packets <- normal
+	time.Sleep(time.Millisecond)
+	priorityPackets <- priority
+
+	deadline := time.After(time.Second)
+	for {
+		if batch := transport.firstBatch(); len(batch) == 1 {
+			if string(batch[0]) != string(priority) {
+				t.Fatalf("first uploaded packet = %#v, want priority packet before pending normal batch", batch[0])
+			}
+			return
+		}
+		select {
+		case <-deadline:
+			t.Fatal("timed out waiting for preempted priority upload batch")
+		default:
+			time.Sleep(time.Millisecond)
+		}
+	}
+}
+
 func TestGatewayUploadMicroBatchesTCPControlPackets(t *testing.T) {
 	transport := &recordingGatewayTransport{}
 	gateway := &Gateway{Transport: transport, VPNConnectionID: "vpn-1"}

clients/android/app/build.gradle

@@ -30,8 +30,8 @@ android {
         applicationId "su.cin.rapvpn"
         minSdk 26
         targetSdk 35
-        versionCode 203
+        versionCode 204
-        versionName "0.2.203"
+        versionName "0.2.204"
         buildConfigField "String", "DEFAULT_BACKEND_URL", "\"${normalizeGradleString(defaultBackendUrl)}\""
         buildConfigField "String", "DEFAULT_CLUSTER_ID", "\"${normalizeGradleString(defaultClusterId)}\""
         buildConfigField "String", "DEFAULT_ORGANIZATION_ID", "\"${normalizeGradleString(defaultOrganizationId)}\""

clients/android/app/src/main/java/su/cin/rapvpn/RapDiagnosticService.java

@@ -91,6 +91,7 @@ public class RapDiagnosticService extends Service {
     private volatile long heartbeatStartedAt = 0;
     private volatile long commandPollStartedAt = 0;
     private volatile long commandStartedAt = 0;
+    private volatile long lastFabricLeaseRefreshAttemptAt = 0;
     private String controlNetworkMode = "";
     private final AtomicBoolean heartbeatInProgress = new AtomicBoolean(false);
     private final AtomicBoolean commandPollInProgress = new AtomicBoolean(false);
@@ -332,6 +333,11 @@ public class RapDiagnosticService extends Service {
                 } catch (Exception e) {
                     serviceState = "upgrade restart check warning: " + e.getMessage();
                 }
+                try {
+                    maybeRecoverExpiredFabricLease();
+                } catch (Exception e) {
+                    serviceState = "fabric lease recovery warning: " + e.getMessage();
+                }
                 lastWorkerProgressAt = System.currentTimeMillis();
                 reportStatusWithFallback(backendUrl, clusterId, deviceId, statusPayload("heartbeat"));
                 lastWorkerProgressAt = System.currentTimeMillis();
@@ -803,7 +809,7 @@ public class RapDiagnosticService extends Service {
         try {
             String refreshToken = new SecureTokenStore(this).get(PREF_REFRESH_TOKEN);
             if (refreshToken.isEmpty()) {
-                String savedUserId = prefs.getString(PREF_USER_ID, "");
+                String savedUserId = savedUserIdForProfileRefresh(prefs);
                 if (savedUserId == null || savedUserId.trim().isEmpty()) {
                     return "refresh_profile skipped: refresh token and saved user missing";
                 }
@@ -818,6 +824,56 @@ public class RapDiagnosticService extends Service {
         }
     }
 
+    private String savedUserIdForProfileRefresh(SharedPreferences prefs) {
+        String savedUserId = prefs.getString(PREF_USER_ID, "");
+        if (savedUserId != null && !savedUserId.trim().isEmpty()) {
+            return savedUserId.trim();
+        }
+        try {
+            String profileJson = prefs.getString(PREF_PROFILE_JSON, "");
+            if (profileJson == null || profileJson.trim().isEmpty()) {
+                return "";
+            }
+            JSONObject root = new JSONObject(profileJson);
+            JSONObject profile = root.optJSONObject("vpn_client_profile");
+            if (profile == null) {
+                profile = root;
+            }
+            String profileUserId = profile.optString("user_id", "").trim();
+            if (!profileUserId.isEmpty()) {
+                prefs.edit().putString(PREF_USER_ID, profileUserId).apply();
+                return profileUserId;
+            }
+        } catch (Exception ignored) {
+        }
+        return "";
+    }
+
+    private void maybeRecoverExpiredFabricLease() {
+        SharedPreferences runtime = getSharedPreferences(RUNTIME_PREFS, MODE_PRIVATE);
+        String downlinkError = runtime.getString("downlink_error_type", "") + " " + runtime.getString("downlink_message", "");
+        String uplinkError = runtime.getString("uplink_sender_error_type", "") + " " + runtime.getString("uplink_sender_message", "");
+        String combined = (downlinkError + " " + uplinkError).toLowerCase();
+        if (!combined.contains("403 forbidden") && !combined.contains("service channel lease expired")) {
+            return;
+        }
+        long now = System.currentTimeMillis();
+        if (now - lastFabricLeaseRefreshAttemptAt < 60000) {
+            return;
+        }
+        lastFabricLeaseRefreshAttemptAt = now;
+        String refresh = refreshProfile();
+        if (!refresh.startsWith("refresh_profile ok")) {
+            serviceState = "fabric lease recovery refresh skipped: " + refresh;
+            return;
+        }
+        String start = startVPNFromSavedProfile();
+        serviceState = "fabric lease recovery restarted vpn: " + start;
+        lastCommandType = "auto_fabric_lease_recovery";
+        lastCommandResult = start + " profile_refresh=" + refresh;
+        lastCommandAt = now;
+    }
+
     private String refreshProfileForUser(SharedPreferences prefs, String userId, String trustedDeviceId) throws Exception {
         String backendUrl = normalizeBackendUrl(prefs.getString("backend_url", DEFAULT_BACKEND_URL));
         String organizationId = prefs.getString("organization_id", DEFAULT_ORGANIZATION_ID);