Fix memory leak in Completable/Single Processors (#2114)

Motivation:
`Processors.new[Single|Completable]Processor` create a `Processor`s
that permit multiple subscribers. The underlying datastructure is
`ClosableConcurrentStack` that makes a best effort to remove items
on cancellation by setting the `Subscriber` reference to `null`.
However it doesn't attempt to remove the `Node` which can lead
to a memory leak until `close(..)` is called. This may not happen
in scenarios that perpetually retry operations that are failing
continuously for prolonged periods of time.

Modifications:
- `ClosableConcurrentStack` makes a best effort to remove the
  `Node` object from the stack in `relaxedRemove`.

Result:
Memory leak fixed.

servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/ClosableConcurrentStack.java

@@ -70,15 +70,27 @@ boolean relaxedRemove(T item) {
             return false;
         }
         @SuppressWarnings("unchecked")
-        Node<T> currTop = (Node<T>) rawCurrTop;
-        while (currTop != null) {
-            if (item.equals(currTop.item)) {
-                currTop.item = null; // best effort null out the item. pop/close will discard the Node later.
+        Node<T> curr = (Node<T>) rawCurrTop;
+        Node<T> prev = null;
+        do {
+            if (item.equals(curr.item)) {
+                // Best effort Node removal. close will discard the Node if this attempt isn't visible to all threads.
+                curr.item = null;
+                if (prev != null) {
+                    prev.next = curr.next;
+                    // Don't set curr.next to null! If multiple threads are removing items at the same time removed
+                    // Nodes maybe "resurrected" and if links are nulled the stack may lose references to Nodes after
+                    // the removal point. Just let the GC reclaim the Node when it is ready because it is no longer
+                    // referenced from top, or if "resurrected" the "relaxed" contract of this method allows it.
+                } else {
+                    topUpdater.compareAndSet(this, curr, curr.next);
+                }
                 return true;
             } else {
-                currTop = currTop.next;
+                prev = curr;
+                curr = curr.next;
             }
-        }
+        } while (curr != null);
         return false;
     }
 

servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/ClosableConcurrentStackTest.java

@@ -33,8 +33,8 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.in;
 import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.isIn;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -101,7 +101,7 @@ void concurrentPushClose() throws Exception {
         future.get();
         assertEquals(itemCount, overallValues.size());
         for (int i = 0; i < itemCount; ++i) {
-            assertThat(i, isIn(overallValues));
+            assertThat(i, is(in(overallValues)));
         }
     }