Ratchet: Expire pending outbound

Prep for Bob transition to ES
Log tweaks

router/java/src/net/i2p/router/crypto/ratchet/RatchetSKM.java

@@ -68,6 +68,8 @@ public class RatchetSKM extends SessionKeyManager implements SessionTagListener
      */
     private final static long SESSION_LIFETIME_MAX_MS = SESSION_TAG_DURATION_MS + 3 * 60 * 1000;
 
+    private final static long SESSION_PENDING_DURATION_MS = 5 * 60 * 1000;
+
     /**
      * Time to send more if we are this close to expiration
      */
@@ -522,22 +524,29 @@ public class RatchetSKM extends SessionKeyManager implements SessionTagListener
                 _log.debug("IB tag not found: " + tag.toBase64());
             return null;
         }
-        HandshakeState state = tagSet.getHandshakeState();
+        boolean firstInbound;
         synchronized(tagSet) {
+            firstInbound = !tagSet.getAcked();
             key = tagSet.consume(tag);
             if (key != null)
                 tagSet.setDate(_context.clock().now());
         }
-        if (key == null) {
+        if (key != null) {
-            if (_log.shouldDebug())
+            HandshakeState state = tagSet.getHandshakeState();
-                _log.debug("tag " + tag + " not found in tagset!!! " + tagSet);
+            if (firstInbound) {
-        }
+                if (state == null) {
-        if (state != null) {
+                    // TODO
-            if (_log.shouldDebug())
+                }
-                _log.debug("IB NSR Tag consumed: " + tag + " from: " + tagSet);
+            }
+            if (_log.shouldDebug()) {
+                if (state != null)
+                    _log.debug("IB NSR Tag consumed: " + tag + " from: " + tagSet);
+                else
+                    _log.debug("IB ES Tag consumed: " + tag + " from: " + tagSet);
+            }
         } else {
-            if (_log.shouldDebug())
+            if (_log.shouldWarn())
-                _log.debug("IB ES Tag consumed: " + tag + " from: " + tagSet);
+                _log.warn("tag " + tag + " not found in tagset!!! " + tagSet);
         }
         return key;
     }
@@ -598,7 +607,27 @@ public class RatchetSKM extends SessionKeyManager implements SessionTagListener
         }
         if (oremoved > 0 && _log.shouldInfo())
             _log.info("Expired outbound: " + oremoved);
-        return removed + oremoved;
+
+        int premoved = 0;
+        exp = now - SESSION_PENDING_DURATION_MS;
+        synchronized (_pendingOutboundSessions) {
+            for (Iterator<List<OutboundSession>> iter = _pendingOutboundSessions.values().iterator(); iter.hasNext();) {
+                List<OutboundSession> pending = iter.next();
+                for (Iterator<OutboundSession> liter = pending.iterator(); liter.hasNext();) {
+                    OutboundSession sess = liter.next();
+                    if (sess.getEstablishedDate() < exp) {
+                        liter.remove();
+                        premoved++;
+                    }
+                }
+                if (pending.isEmpty())
+                    iter.remove();
+            }
+        }
+        if (premoved > 0 && _log.shouldInfo())
+            _log.info("Expired pending: " + premoved);
+
+        return removed + oremoved + premoved;
     }
 
     /// begin SessionTagListener ///

router/java/src/net/i2p/router/crypto/ratchet/RatchetTagSet.java

@@ -15,6 +15,7 @@ import net.i2p.crypto.TagSetHandle;
 import net.i2p.data.Base64;
 import net.i2p.data.DataHelper;
 import net.i2p.data.SessionKey;
+import net.i2p.util.Log;
 
 /**
  *  A tagset class for one direction, either inbound or outbound.
@@ -256,10 +257,13 @@ class RatchetTagSet implements TagSetHandle {
         // == not equals
         int idx = _sessionTags.indexOfValueByValue(tag);
         if (idx < 0) {
-            System.out.println("Tag not found " + Base64.encode(tag.getData()));
+            Log log = I2PAppContext.getGlobalContext().logManager().getLog(RatchetTagSet.class);
-            System.out.println("Remaining tags: " + getTags());
+            if (log.shouldWarn())
+                log.warn("Tag not found " + Base64.encode(tag.getData()) +
+                         " Remaining tags: " + getTags(), new Exception());
             return null;
         }
+        _acked = true;
         int tagnum = _sessionTags.keyAt(idx);
         _sessionTags.removeAt(idx);
 
@@ -289,8 +293,10 @@ class RatchetTagSet implements TagSetHandle {
             return rv;
         } else {
             // dup or some other error
-            System.out.println("No key found for tag " + Base64.encode(tag.getData()) + " at index " + idx +
+            Log log = I2PAppContext.getGlobalContext().logManager().getLog(RatchetTagSet.class);
-                               " tagnum = " + tagnum + " lastkey = " + _lastKey);
+            if (log.shouldWarn())
+                log.warn("No key found for tag " + Base64.encode(tag.getData()) + " at index " + idx +
+                         " tagnum = " + tagnum + " lastkey = " + _lastKey, new Exception());
             return null;
         }
     }
@@ -309,7 +315,7 @@ class RatchetTagSet implements TagSetHandle {
         // trim if too big
         int toTrim = _sessionTags.size() - _maxSize;
         if (toTrim > 0) {
-            System.out.println("Trimming tags by " + toTrim);
+            //System.out.println("Trimming tags by " + toTrim);
             for (int i = 0; i < toTrim; i++) {
                 int tagnum = _sessionTags.keyAt(i);
                 int kidx = _sessionKeys.indexOfKey(tagnum);
@@ -362,12 +368,13 @@ class RatchetTagSet implements TagSetHandle {
     }
 
     /**
-     *  For outbound only.
+     *  For outbound only, call when we can use it.
      */
     public void setAcked() { _acked = true; }
 
     /**
-     *  For outbound only.
+     *  For inbound, returns true after first consume() call.
+     *  For outbound, returns true after set.
      */
     public boolean getAcked() { return _acked; }