Merge pull request #1331 from myronkscott/harden_client_ref

Make the server more tolerant of client mistakes during reference/dereference

Author: Myron Scott

tc-client/src/main/java/com/tc/object/ClientEntityManagerImpl.java

@@ -57,7 +57,6 @@
 import com.tc.text.MapListPrettyPrint;
 import com.tc.text.PrettyPrintable;
 import com.tc.util.Assert;
-import com.tc.util.Util;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
@@ -80,6 +79,7 @@
 
 import static com.tc.object.EntityDescriptor.createDescriptorForLifecycle;
 import static com.tc.object.SafeInvocationCallback.safe;
+import java.util.concurrent.Callable;
 import static java.util.stream.Collectors.toCollection;
 import static org.terracotta.entity.Invocation.uninterruptiblyGet;
 
@@ -409,6 +409,20 @@ private <M extends EntityMessage, R extends EntityResponse> EntityClientEndpoint
     Assert.assertNotNull("Can't lookup null entity descriptor", instance);
     final EntityDescriptor fetchDescriptor = EntityDescriptor.createDescriptorForFetch(entity, version, instance);
     EntityClientEndpointImpl<M, R> resolvedEndpoint = null;
+    // make sure release is only ever called once
+    Callable<Void> closeCall = new Callable<Void>() {
+      boolean released = false;
+      
+      @Override
+      public synchronized Void call() throws Exception {
+        if (!released) {
+          released = true;
+          internalRelease(entity, fetchDescriptor);
+        }
+        return null;
+      }
+    };
+    
     try {
       byte[] raw = internalRetrieve(fetchDescriptor);
       ByteBuffer br = ByteBuffer.wrap(raw);
@@ -418,21 +432,8 @@ private <M extends EntityMessage, R extends EntityResponse> EntityClientEndpoint
       br.get(config);
       // We can only fail to get the config if we threw an exception.
       Assert.assertTrue(null != raw);
-      // We managed to retrieve the config so create the end-point.
-      // Note that we will need to call release on this descriptor, when it is closed, prior to running the closeHook we
-      //  were given so combine these ideas to pass in one runnable.
-      Runnable compoundRunnable = new Runnable() {
-        @Override
-        public void run() {
-          try {
-            internalRelease(entity, fetchDescriptor);
-          } catch (EntityException e) {
-            // We aren't expecting there to be any problems releasing an entity in the close hook so we will just log and re-throw.
-            Util.printLogAndRethrowError(e, logger);
-          }
-        }
-      };
-      resolvedEndpoint = new EntityClientEndpointImpl<>(entity, version, EntityDescriptor.createDescriptorForInvoke(fetch, instance), this, config, codec, compoundRunnable, this.endpointCloser);
+
+      resolvedEndpoint = new EntityClientEndpointImpl<>(entity, version, EntityDescriptor.createDescriptorForInvoke(fetch, instance), this, config, codec, closeCall, this.endpointCloser);
 
       if (this.objectStoreMap.putIfAbsent(instance, resolvedEndpoint) != null) {
         throw Assert.failure("Attempt to add an object that already exists: Object of class " + resolvedEndpoint.getClass()
@@ -442,14 +443,22 @@ public void run() {
       throw notfound;
     } catch (EntityException e) {
       // Release the entity and re-throw to the higher level.
-      internalRelease(entity, fetchDescriptor);
-      // NOTE:  Since we are throwing, we are not responsible for calling the given closeHook.
+      try {
+        closeCall.call();
+      } catch (Exception subE) {
+        e.addSuppressed(subE);
+      }
+
       throw e;
     } catch (Throwable t) {
       // This is the unexpected case so clean up and re-throw as a RuntimeException
       // Clean up any client-side or server-side state regarding this failed connection.
-      internalRelease(entity, fetchDescriptor);
-      // NOTE:  Since we are throwing, we are not responsible for calling the given closeHook.
+      try {
+        closeCall.call();
+      } catch (Exception subE) {
+        t.addSuppressed(subE);
+      }
+
       throw Throwables.propagate(t);
     }
 

tc-client/src/main/java/com/tc/object/EntityClientEndpointImpl.java

@@ -40,6 +40,7 @@
 import org.slf4j.LoggerFactory;
 
 import static com.tc.object.SafeInvocationCallback.safe;
+import org.terracotta.exception.EntityException;
 
 
 public class EntityClientEndpointImpl<M extends EntityMessage, R extends EntityResponse> implements EntityClientEndpoint<M, R> {
@@ -52,7 +53,7 @@ public class EntityClientEndpointImpl<M extends EntityMessage, R extends EntityR
   private final EntityID entityID;
   private final long version;
   private final MessageCodec<M, R> codec;
-  private final Runnable closeHook;
+  private final Callable<Void> closeHook;
   private final ExecutorService closer;
   private EndpointDelegate<R> delegate;
   private boolean isOpen;
@@ -66,7 +67,7 @@ public class EntityClientEndpointImpl<M extends EntityMessage, R extends EntityR
    * @param entityConfiguration Opaque byte[] describing how to configure the entity to be built on top of this end-point.
    * @param closeHook A Runnable which will be run last when the end-point is closed.
    */
-  public EntityClientEndpointImpl(EntityID eid, long version, EntityDescriptor instance, InvocationHandler invocationHandler, byte[] entityConfiguration, MessageCodec<M, R> codec, Runnable closeHook, ExecutorService closer) {
+  public EntityClientEndpointImpl(EntityID eid, long version, EntityDescriptor instance, InvocationHandler invocationHandler, byte[] entityConfiguration, MessageCodec<M, R> codec, Callable<Void> closeHook, ExecutorService closer) {
     this.entityID = eid;
     this.version = version;
     this.invokeDescriptor = instance;
@@ -176,12 +177,16 @@ public byte[] getExtendedReconnectData() {
   @Override
   public void close() {
     // We can't close twice.
-    checkEndpointOpen();
-    if (this.closeHook != null) {
-      this.closeHook.run();
+    if (closeIfOpen()) {
+      if (this.closeHook != null) {
+        try {
+          this.closeHook.call();
+        } catch (Exception e) {
+          LOGGER.warn("Exception occured during close", e);
+        }
+        // log and swallow this exception closing
+      }
     }
-    // We also need to invalidate ourselves so we don't continue allowing new messages through when disconnecting.
-    this.isOpen = false;
   }
 
   @Override
@@ -220,9 +225,15 @@ public void didCloseUnexpectedly() {
     }
   }
 
-  private void checkEndpointOpen() {
+  private synchronized void checkEndpointOpen() {
     if (!this.isOpen) {
       throw new IllegalStateException("Endpoint closed");
     }
   }
+
+  private synchronized boolean closeIfOpen() {
+    boolean wasOpen = this.isOpen;
+    this.isOpen = false;
+    return wasOpen;
+  }
 }

tc-server/src/main/java/com/tc/objectserver/entity/ManagedEntityImpl.java

@@ -976,16 +976,27 @@ private void getEntity(ServerEntityRequest getEntityRequest, ResultCapture respo
     if (this.isDestroyed) {
       response.failure(ServerException.createNotFoundException(getID()));
     } else {
-      if (canDelete) {
-        clientReferenceCount += 1;
-        Assert.assertTrue(clientReferenceCount > 0);
-      }
       // The FETCH can only come directly from a client so we can down-cast.
       ClientID clientID = getEntityRequest.getNodeID();
       ClientDescriptorImpl descriptor = new ClientDescriptorImpl(clientID, getEntityRequest.getClientInstance());
       boolean added = clientEntityStateManager.addReference(descriptor, this.fetchID);
+      
+      if (canDelete) {
+        if (added) {
+          clientReferenceCount += 1;
+          Assert.assertTrue(clientReferenceCount > 0);
+        } else {
+          // this should never happen.  Ther server is expecting sane things from the client. 
+          logger.warn("the client has attempted to fetch the same entity instance twice {}", descriptor);
+        }
+      }
+      
       if (this.isInActiveState) {
-        Assert.assertTrue(added);
+        if (!added) {
+          //  Exception the client.  Don't crash the server
+          response.failure(ServerException.createReferencedException(id));
+          return;
+        }
         // Fire the event that the client fetched the entity.
         this.eventCollector.clientDidFetchEntity(clientID, this.id, this.consumerID, getEntityRequest.getClientInstance());
         // finally notify the entity that it was fetched
@@ -1006,10 +1017,18 @@ private void getEntity(ServerEntityRequest getEntityRequest, ResultCapture respo
               handler.handleReconnect(descriptor, extendedData);
             }
           } catch (ReconnectRejectedException rejected) {
+            Assert.assertTrue(clientEntityStateManager.removeReference(descriptor));
+            if (canDelete) {
+              clientReferenceCount -= 1;
+            }
             response.failure(ServerException.createReconnectRejected(getID(), rejected));
             return;
           } catch (Exception e) {
 //  something happened during reconnection, force a disconnection, see ProcessTransactionHandler.disconnectClientDueToFailure for handling
+            Assert.assertTrue(clientEntityStateManager.removeReference(descriptor));
+            if (canDelete) {
+              clientReferenceCount -= 1;
+            }
             logger.warn("unexpected exception.  rejecting reconnection of " + descriptor.getNodeID() + " to " + this.id, e);
             response.failure(ServerException.createReconnectRejected(getID(), new ReconnectRejectedException(e.getMessage(), e)));
             return;
@@ -1031,17 +1050,27 @@ private void releaseEntity(ServerEntityRequest request, ResultCapture response)
     if (this.isDestroyed) {
       response.failure(ServerException.createNotFoundException(this.getID()));
     } else {
-      if (canDelete) {
-        clientReferenceCount -= 1;
-        Assert.assertTrue(clientReferenceCount >= 0);
-      }
-
       ClientID clientID = request.getNodeID();
       ClientDescriptorImpl clientInstance = new ClientDescriptorImpl(clientID, request.getClientInstance());
       boolean removed = clientEntityStateManager.removeReference(clientInstance);
 
+      if (canDelete) {
+        if (removed) {
+          clientReferenceCount -= 1;
+          Assert.assertTrue(clientReferenceCount >= 0);
+        } else {
+        // this should never happen.  the expectation is that the client will send sane things to the server.
+          logger.warn("client has tried to release an entity that is not fetched {}", clientInstance);
+        }
+      }
+
       if (this.isInActiveState) {
-        Assert.assertTrue(removed);
+        if (!removed) {
+          //  Exception the client.  don't crash the server
+          response.failure(ServerException.createNotFoundException(id));
+          return;
+        }
+        
         this.activeServerEntity.disconnected(clientInstance);
         // Fire the event that the client released the entity.
         this.eventCollector.clientDidReleaseEntity(clientID, this.id, this.consumerID, request.getClientInstance());
@@ -1077,6 +1106,8 @@ public Runnable promoteEntity() throws ConfigurationException {
 // up on passives
     if (canDelete) {
       this.clientReferenceCount = 0;
+      // the entity references should have been cleared
+      Assert.assertTrue(clientEntityStateManager.verifyNoEntityReferences(fetchID));
     } else {
       Assert.assertEquals(this.clientReferenceCount, ManagedEntityImpl.UNDELETABLE_ENTITY);
     }

tc-server/src/test/java/com/tc/objectserver/entity/ManagedEntityImplTest.java

@@ -173,6 +173,7 @@ public void setUp() throws Exception {
     messageSelf = mock(Sink.class);
     when(clientEntityStateManager.addReference(any(ClientDescriptorImpl.class), any(FetchID.class))).thenReturn(Boolean.TRUE);
     when(clientEntityStateManager.removeReference(any(ClientDescriptorImpl.class))).thenReturn(Boolean.TRUE);
+    when(clientEntityStateManager.verifyNoEntityReferences(any())).thenReturn(Boolean.TRUE);
     eventCollector = new ManagementTopologyEventCollector(mock(IMonitoringProducer.class));
     // We will start this in a passive state, as the general test case.
     boolean isInActiveState = false;