CBL-8071: Do not retry replicator connections if WebSocket being closed abnormally

Networking/WebSockets/WebSocketImpl.cc

@@ -539,7 +539,8 @@ namespace litecore::websocket {
 
                     if ( clean ) {
                         status.reason = kWebSocketClose;
-                        if ( !expected ) status.code = kCodeAbnormal;
+                        // If !expected, it follows that _closeSent implies !_closeReceived
+                        if ( !expected ) status.code = _closeSent ? kCodeNormal : kCodeAbnormal;
                         else if ( !_closeMessage )
                             status.code = kCodeNormal;
                         else {

Replicator/tests/ReplicatorAPITest.cc

@@ -828,6 +828,86 @@ TEST_CASE_METHOD(ReplicatorAPITest, "Stop after transient connect failure", "[C]
     waitForStatus(kC4Stopped);
 }
 
+// CBL-8074
+TEST_CASE_METHOD(ReplicatorAPITest, "WebSocket Peer Going Away", "[C][Push][Pull]") {
+    bool            afterClose = false;
+    C4SocketFactory factory    = {};
+    C4Socket*       c4socket   = nullptr;
+    factory.context            = &c4socket;
+    factory.open               = [](C4Socket* socket, const C4Address* addr, C4Slice options, void* context) {
+        c4socket_opened(socket);
+        *(C4Socket**)context = socket;
+    };
+
+    factory.close = [](C4Socket* socket) {
+        // Not invoked
+        REQUIRE(false);
+    };
+
+    // "peer going away" before CLOSE is sent
+    // Replicator receives error code 1006, which is transient.
+    // C4Replicator goes to offline and waiting for retry.
+    SECTION("CLOSE Not Sent") {
+        afterClose    = false;
+        _mayGoOffline = true;
+        factory.write = [](C4Socket* socket, C4SliceResult msg) {
+            // Simulate Peer-Going-Away before Replicator calling Stop.
+            // Socket is closed unexpectedly, without the client sending CLOSE
+            FLSliceResult_Release(msg);
+            c4socket_closed(socket, {WebSocketDomain, websocket::kCodeGoingAway});
+        };
+    }
+
+    // "peer going away" after CLOSE frame was already sent
+    // Since the replicator is already stopped when the peer goes away, WebSocket will
+    // treat it as Normal Close.
+    SECTION("CLOSE Has Been Sent") {
+        afterClose    = true;
+        _mayGoOffline = false;
+        factory.write = [](C4Socket* socket, C4SliceResult msg) {
+            // Do nothing
+            FLSliceResult_Release(msg);
+        };
+    }
+
+    _socketFactory = &factory;
+    C4Error err;
+    importJSONLines(sFixturesDir + "names_100.json");
+
+    if ( !afterClose ) {
+        // WebSocket code 1006, transient error
+        REQUIRE(startReplicator(kC4Disabled, kC4OneShot, WITH_ERROR(&err)));
+        _numCallbacksWithLevel[kC4Offline] = 0;
+        waitForStatus(kC4Offline);
+    } else {
+        REQUIRE(startReplicator(kC4Disabled, kC4Continuous, WITH_ERROR(&err)));
+        // Making sure the WebSocket is open/connected
+        waitForStatus(kC4Busy);
+    }
+
+    c4repl_stop(_repl);
+
+    if ( afterClose ) {
+        // Give some time for Replicator::_stop to be called, but before timeout in WebSocketImpl
+        // to not get Timeout error.
+        std::this_thread::sleep_for(1s);
+        // WebSocket will treat it as Normal Close
+        c4socket_closed(c4socket, {WebSocketDomain, websocket::kCodeGoingAway});
+    }
+
+    waitForStatus(kC4Stopped);
+
+    auto status = c4repl_getStatus(_repl);
+
+    if ( !afterClose ) {
+        // kCodeAbnormal == 1006
+        CHECK((status.error.domain == WebSocketDomain && status.error.code == websocket::kCodeAbnormal));
+    } else {
+        // "peer going away" after stop results in normal Stop.
+        CHECK(status.error.code == 0);
+    }
+}
+
 TEST_CASE_METHOD(ReplicatorAPITest, "Calling c4socket_ method after STOP", "[C][Push][Pull]") {
     // c.f. the flow with test case "Stop after transient connect failure"
     _mayGoOffline            = true;