ZOOKEEPER-4689: Fix ACL cache recovery from fuzzy snapshots

zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java

@@ -96,6 +96,11 @@ public class DataTree {
 
     private static final Logger LOG = LoggerFactory.getLogger(DataTree.class);
 
+    // Deny watch delivery to ordinary sessions when a node's ACL is unknown.
+    // checkACL treats null and empty ACLs as unrestricted. Public for tests.
+    public static final List<ACL> NO_PERMISSIONS_ACL =
+        Collections.singletonList(new ACL(0, ZooDefs.Ids.ANYONE_ID_UNSAFE));
+
     private final RateLogger RATE_LOGGER = new RateLogger(LOG, 15 * 60 * 1000);
 
     /**
@@ -441,24 +446,26 @@ public void createNode(final String path, byte[] data, List<ACL> acl, long ephem
         }
         List<ACL> parentAcl;
         synchronized (parent) {
-            parentAcl = getACL(parent);
+            parentAcl = getACLForWatchTrigger(parent, parentName);
 
-            // Add the ACL to ACL cache first, to avoid the ACL not being
-            // created race condition during fuzzy snapshot sync.
-            //
-            // This is the simplest fix, which may add ACL reference count
-            // again if it's already counted in the ACL map of fuzzy
-            // snapshot, which might also happen for deleteNode txn, but
-            // at least it won't cause the ACL not exist issue.
-            //
-            // Later we can audit and delete all non-referenced ACLs from
-            // ACL map when loading the snapshot/txns from disk, like what
-            // we did for the global sessions.
-            Long acls = aclCache.convertAcls(acl);
+            // Intern the ACL before checking whether the node exists: a fuzzy
+            // snapshot can contain the node but omit its ACL (ZOOKEEPER-3306).
+            Long aclId = aclCache.convertAcls(acl);
 
             DataNode existingChild = nodes.get(path);
             if (existingChild != null) {
-                existingChild.acl = acls;
+                // Use the ACL value from the txn; snapshot ids are server-local
+                // and may not resolve. Release the node's old reference to balance
+                // convertAcls, including when both ids are equal.
+                synchronized (existingChild) {
+                    Long previousAcl = existingChild.acl;
+                    existingChild.acl = aclId;
+                    aclCache.removeUsage(previousAcl);
+                    if (!aclId.equals(previousAcl)) {
+                        LOG.info("ACL reference of existing node {} changed from {} to {} while applying a create txn",
+                            path, previousAcl, aclId);
+                    }
+                }
                 throw new NodeExistsException();
             }
 
@@ -476,7 +483,7 @@ public void createNode(final String path, byte[] data, List<ACL> acl, long ephem
                 parent.stat.setCversion(parentCVersion);
                 parent.stat.setPzxid(zxid);
             }
-            DataNode child = new DataNode(data, acls, stat);
+            DataNode child = new DataNode(data, aclId, stat);
             parent.addChild(childName);
             nodes.postChange(parentName, parent);
             nodeDataSize.addAndGet(getNodeSize(path, child.data));
@@ -562,7 +569,7 @@ public void deleteNode(String path, long zxid) throws NoNodeException {
         List<ACL> acl;
         nodes.remove(path);
         synchronized (node) {
-            acl = getACL(node);
+            acl = getACLForWatchTrigger(node, path);
             aclCache.removeUsage(node.acl);
             nodeDataSize.addAndGet(-getNodeSize(path, node.data));
         }
@@ -572,7 +579,7 @@ public void deleteNode(String path, long zxid) throws NoNodeException {
         // separate patch.
         List<ACL> parentAcl;
         synchronized (parent) {
-            parentAcl = getACL(parent);
+            parentAcl = getACLForWatchTrigger(parent, parentName);
             long owner = node.stat.getEphemeralOwner();
             EphemeralType ephemeralType = EphemeralType.get(owner);
             if (ephemeralType == EphemeralType.CONTAINER) {
@@ -635,7 +642,7 @@ public Stat setData(String path, byte[] data, int version, long zxid, long time)
         List<ACL> acl;
         byte[] lastData;
         synchronized (n) {
-            acl = getACL(n);
+            acl = getACLForWatchTrigger(n, path);
             lastData = n.data;
             nodes.preChange(path, n);
             n.data = data;
@@ -792,6 +799,27 @@ public List<ACL> getACL(DataNode node) {
         }
     }
 
+    /**
+     * Returns the ACL for filtering a node's watch events. A node loaded from a
+     * fuzzy snapshot can reference an ACL id that is missing from the cache
+     * until replay repairs it. In that case this returns
+     * {@link #NO_PERMISSIONS_ACL} instead of throwing, so replay can continue.
+     */
+    private List<ACL> getACLForWatchTrigger(DataNode node, String path) {
+        Long aclId;
+        List<ACL> acls;
+        synchronized (node) {
+            aclId = node.acl;
+            acls = aclCache.convertLongIfCached(aclId);
+        }
+        if (acls != null) {
+            return acls;
+        }
+        LOG.info("ACL id {} of {} is not in the ACL cache; filtering its watch events as unreadable",
+            aclId, path);
+        return NO_PERMISSIONS_ACL;
+    }
+
     public int aclCacheSize() {
         return aclCache.size();
     }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ReferenceCountedACLCache.java

@@ -95,6 +95,24 @@ public synchronized List<ACL> convertLong(Long longVal) {
         return acls;
     }
 
+    /**
+     * Converts a long to a list of ACLs without throwing if it is absent from
+     * the cache. A node in a fuzzy snapshot can temporarily reference such an
+     * absent id until transaction replay repairs it (ZOOKEEPER-4689).
+     *
+     * @param longVal ACL id
+     * @return the corresponding ACLs, or null if the id is not cached
+     */
+    synchronized List<ACL> convertLongIfCached(Long longVal) {
+        if (longVal == null) {
+            return null;
+        }
+        if (longVal == OPEN_UNSAFE_ACL_ID) {
+            return ZooDefs.Ids.OPEN_ACL_UNSAFE;
+        }
+        return longKeyMap.get(longVal);
+    }
+
     private long incrementIndex() {
         return ++aclIndex;
     }
@@ -170,6 +188,15 @@ public synchronized void addUsage(Long acl) {
             return;
         }
 
+        if (acl > aclIndex) {
+            // The ACL cache is serialized before the nodes in a fuzzy snapshot.
+            // A node can therefore reference an id absent from the serialized
+            // cache. Reserve every id mentioned by the loaded tree so replay
+            // cannot assign it to a different ACL before repairing the node.
+            LOG.info("Advancing aclIndex from {} to {} based on a node reference in the snapshot", aclIndex, acl);
+            aclIndex = acl;
+        }
+
         if (!longKeyMap.containsKey(acl)) {
             LOG.info("Ignoring acl {} as it does not exist in the cache", acl);
             return;

zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java

@@ -33,6 +33,8 @@
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Semaphore;
@@ -42,15 +44,18 @@
 import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.InputArchive;
 import org.apache.jute.Record;
+import org.apache.zookeeper.KeeperException.NoAuthException;
 import org.apache.zookeeper.KeeperException.NoNodeException;
 import org.apache.zookeeper.KeeperException.NodeExistsException;
 import org.apache.zookeeper.Quotas;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.common.PathTrie;
+import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.metrics.MetricsUtils;
 import org.apache.zookeeper.txn.CreateTxn;
+import org.apache.zookeeper.txn.DeleteTxn;
 import org.apache.zookeeper.txn.TxnHeader;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
@@ -647,6 +652,49 @@ public void testCreateNodeFixMissingACL() throws Exception {
         dt.createNode("/the_parent/the_child", new byte[0], ZooDefs.Ids.CREATOR_ALL_ACL, -1, 2, 2, 2);
     }
 
+    /**
+     * Applying a create txn to an existing node (routine when replay overlaps
+     * a fuzzy snapshot, e.g. after a restart or SNAP sync) must keep the ACL
+     * reference counts exact, so that unused ACLs are garbage collected and
+     * the cache does not grow until the next reload.
+     *
+     * See ZOOKEEPER-4689.
+     */
+    @Test
+    public void testCreateTxnReplayOnExistingNodeKeepsAclRefCountExact() throws Exception {
+        DataTree dt = new DataTree();
+        int baseline = dt.aclCacheSize();
+
+        TxnHeader hdr = new TxnHeader(1, 2, 2, 2, ZooDefs.OpCode.create);
+        Record txn = new CreateTxn("/x", new byte[0], ZooDefs.Ids.CREATOR_ALL_ACL, false, -1);
+        dt.processTxn(hdr, txn);
+        // Duplicate apply of the same create txn hits the node-exists branch.
+        dt.processTxn(hdr, txn);
+        assertEquals(baseline + 1, dt.aclCacheSize());
+
+        TxnHeader deleteHdr = new TxnHeader(1, 2, 3, 2, ZooDefs.OpCode.delete);
+        Record deleteTxn = new DeleteTxn("/x");
+        dt.processTxn(deleteHdr, deleteTxn);
+        // The ACL is no longer referenced by any node, so it must have been
+        // garbage collected.
+        assertEquals(baseline, dt.aclCacheSize());
+    }
+
+    /**
+     * The ACL used to filter watch events of a node whose ACL is unknown must
+     * deny READ to ordinary sessions. Null or an empty ACL would not, because
+     * checkACL treats both as unrestricted.
+     *
+     * See ZOOKEEPER-4689.
+     */
+    @Test
+    public void testNoPermissionsAclDeniesRead() {
+        ZooKeeperServer zks = new ZooKeeperServer();
+        List<Id> authInfo = Collections.singletonList(new Id("ip", "127.0.0.1"));
+        assertThrows(NoAuthException.class, () ->
+            zks.checkACL(null, DataTree.NO_PERMISSIONS_ACL, ZooDefs.Perms.READ, authInfo, "/x", null));
+    }
+
     private DataTree buildDataTreeForTest() {
         final DataTree dt = new DataTree();
         assertEquals(dt.lastProcessedZxid, 0);