[#1080] Give back the send-window permit of an update the writer drops, and count only what it sends

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java

@@ -980,6 +980,13 @@ public void shutdown()
   /**
    * Select the next update that must be sent to the server managed by this
    * ServerHandler.
+   * <p>
+   * The update comes with a permit of the send window of the session, taken
+   * before the writer decides whether it is sent at all. The peer gives the
+   * permit back for an update it receives, so the writer gives it back itself
+   * for one it drops - see {@link #releasePermitInSendWindow()} - and counts
+   * an update as sent only once it publishes it - see
+   * {@link #countSentUpdate(UpdateMsg)}.
    *
    * @return the next update that must be sent to the server managed by this
    *         ServerHandler.
@@ -1014,16 +1021,46 @@ public UpdateMsg take() throws ChangelogException
       {
         msg = toNotAssuredUpdateMsg(msg);
       }
-      incrementOutCount();
-      if (msg.isAssured())
-      {
-        incrementAssuredStats(msg);
-      }
       return msg;
     }
     return null;
   }
 
+  /**
+   * Gives back the permit of the send window {@link #take()} took for an
+   * update the writer drops rather than sends.
+   * <p>
+   * Only the peer gives permits back, for the updates it receives, and it
+   * never receives this one: kept, the permit would be lost for the rest of
+   * the session, and once more than half of the window is lost that way, the
+   * peer can no longer be sent enough to give any credit back, and the writer
+   * waits for it until the session is re-established (issue #1080).
+   * <p>
+   * A writer which is being shut down is let go of its wait without a permit,
+   * so one given back then may be one it never took: the window of a session
+   * which is going away is not used again.
+   */
+  void releasePermitInSendWindow()
+  {
+    sendWindow.release();
+  }
+
+  /**
+   * Counts an update {@link #take()} returned as sent to the peer, once the
+   * writer publishes it rather than drops it.
+   *
+   * @param msg
+   *          the update the writer publishes
+   */
+  void countSentUpdate(UpdateMsg msg)
+  {
+    incrementOutCount();
+    if (msg.isAssured())
+    {
+      incrementAssuredStats(msg);
+    }
+  }
+
   /**
    * Substitutes a not assured version of the provided update message so that a
    * peer not expected to acknowledge it does not receive it with the assured

opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java

@@ -115,21 +115,30 @@ public void run()
            * ReplicationServerDomain.put() applied when it queued the message, so a peer RS can
            * be given a message this drops - the shutdown must stop waiting for a forward which
            * will never be reported.
+           * <p>
+           * The permit take() charged it is given back as well, whichever filter dropped it: the
+           * server never receives it, so no credit will ever come for it, and nor is it counted
+           * as sent (issue #1080).
            */
+          handler.releasePermitInSendWindow();
           if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
           {
             dsrsShutdownSync.replicaOfflineMsgNotForwarded(
                 replicationServerDomain.getBaseDN(), handler.getServerId());
           }
         }
-        else if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
-        {
-          forwardReplicaOfflineMsg((ReplicaOfflineMsg) updateMsg);
-        }
         else
         {
-          // Publish the update to the remote server using a protocol version it supports
-          session.publish(updateMsg);
+          handler.countSentUpdate(updateMsg);
+          if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
+          {
+            forwardReplicaOfflineMsg((ReplicaOfflineMsg) updateMsg);
+          }
+          else
+          {
+            // Publish the update to the remote server using a protocol version it supports
+            session.publish(updateMsg);
+          }
         }
       }
     }