MNT-24503 - Limits on FixedAclUpdater (#2788)

* Limit the job execution to a configurable amount of nodes (system.fixedACLsUpdater.maxItems)
    * Add query template select_NodesWithAspectIds_Limited that does a left join with alf_store and limits the results
    * Query executions by the job are now limited to maxItemBatchSize
    * Imposing the order by in the query is now configurable (system.fixedACLsUpdater.orderNodes)

repository/src/main/java/org/alfresco/ibatis/IdsEntity.java

@@ -41,6 +41,8 @@ public class IdsEntity
     private Long idFour;
     private List<Long> ids;
     private boolean ordered;
+    private Integer maxResults;
+
     public Long getIdOne()
     {
         return idOne;
@@ -89,4 +91,12 @@ public class IdsEntity
     {
         this.ordered = ordered;
     }
+    public int getMaxResults()
+    {
+        return maxResults;
+    }
+    public void setMaxResults(Integer maxResults)
+    {
+        this.maxResults = maxResults;
+    }
 }

repository/src/main/java/org/alfresco/repo/domain/node/AbstractNodeDAOImpl.java

@@ -2785,6 +2785,23 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO
         selectNodesWithAspects(qnameIds, minNodeId, maxNodeId, ordered, resultsCallback);
     }
 
+    @Override
+    public void getNodesWithAspects(
+            Set<QName> aspectQNames,
+            Long minNodeId, Long maxNodeId, boolean ordered,
+            int maxResults,
+            NodeRefQueryCallback resultsCallback)
+    {
+        Set<Long> qnameIdsSet = qnameDAO.convertQNamesToIds(aspectQNames, false);
+        if (qnameIdsSet.isEmpty())
+        {
+            // No point running a query
+            return;
+        }
+        List<Long> qnameIds = new ArrayList<>(qnameIdsSet);
+        selectNodesWithAspects(qnameIds, minNodeId, maxNodeId, ordered, maxResults, resultsCallback);
+    }
+
     /**
      * @return              Returns a writable copy of the cached aspects set
      */
@@ -4960,6 +4977,10 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO
             List<Long> qnameIds,
             Long minNodeId, Long maxNodeId, boolean ordered,
             NodeRefQueryCallback resultsCallback);
+    protected abstract void selectNodesWithAspects(
+            List<Long> qnameIds,
+            Long minNodeId, Long maxNodeId, boolean ordered, int maxResults,
+            NodeRefQueryCallback resultsCallback);
     protected abstract Long insertNodeAssoc(Long sourceNodeId, Long targetNodeId, Long assocTypeQNameId, int assocIndex);
     protected abstract int updateNodeAssoc(Long id, int assocIndex);
     protected abstract int deleteNodeAssoc(Long sourceNodeId, Long targetNodeId, Long assocTypeQNameId);
@@ -5088,4 +5109,5 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO
     protected abstract Long selectMinTxInNodeIdRange(Long fromNodeId, Long toNodeId);
     protected abstract Long selectMaxTxInNodeIdRange(Long fromNodeId, Long toNodeId);
     protected abstract Long selectNextTxCommitTime(Long fromCommitTime);
+
 }

repository/src/main/java/org/alfresco/repo/domain/node/NodeDAO.java

@@ -427,6 +427,22 @@ public interface NodeDAO extends NodeBulkLoader
             Long minNodeId, Long maxNodeId, boolean ordered,
             NodeRefQueryCallback resultsCallback);
 
+    /**
+     * Get nodes with aspects between the given ranges, ordering the results optionally
+     * and limit the result set
+     *
+     * @param aspectQNames              the aspects that must be on the nodes
+     * @param minNodeId                 the minimum node ID (inclusive)
+     * @param maxNodeId                 the maximum node ID (exclusive)
+     * @param ordered                   if the results are to be ordered by nodeID
+     * @param maxResults                limit query to maxResults
+     * @param resultsCallback           callback to process results
+     */
+    public void getNodesWithAspects(
+            Set<QName> aspectQNames,
+            Long minNodeId, Long maxNodeId, boolean ordered, int maxResults,
+            NodeRefQueryCallback resultsCallback);
+
     /*
      * Node Assocs
      */

repository/src/main/java/org/alfresco/repo/domain/node/ibatis/NodeDAOImpl.java

@@ -116,6 +116,7 @@ public class NodeDAOImpl extends AbstractNodeDAOImpl
     private static final String SELECT_NODE_MAX_ID = "alfresco.node.select_NodeMaxId";
     private static final String SELECT_NODE_INTERVAL_BY_TYPE = "alfresco.node.select_MinMaxNodeIdForNodeType";
     private static final String SELECT_NODES_WITH_ASPECT_IDS = "alfresco.node.select_NodesWithAspectIds";
+    private static final String SELECT_NODES_WITH_ASPECT_IDS_LIMITED = "alfresco.node.select_NodesWithAspectIds_Limited";
     private static final String INSERT_NODE_ASSOC = "alfresco.node.insert.insert_NodeAssoc";
     private static final String UPDATE_NODE_ASSOC = "alfresco.node.update_NodeAssoc";
     private static final String DELETE_NODE_ASSOC = "alfresco.node.delete_NodeAssoc";
@@ -799,6 +800,33 @@ public class NodeDAOImpl extends AbstractNodeDAOImpl
         template.select(SELECT_NODES_WITH_ASPECT_IDS, parameters, resultHandler);
     }
 
+    @Override
+    protected void selectNodesWithAspects(
+            List<Long> qnameIds,
+            Long minNodeId, Long maxNodeId, boolean ordered,
+            final int maxResults,
+            final NodeRefQueryCallback resultsCallback)
+    {
+        @SuppressWarnings("rawtypes")
+        ResultHandler resultHandler = new ResultHandler()
+        {
+            public void handleResult(ResultContext context)
+            {
+                NodeEntity entity = (NodeEntity) context.getResultObject();
+                Pair<Long, NodeRef> nodePair = new Pair<>(entity.getId(), entity.getNodeRef());
+                resultsCallback.handle(nodePair);
+            }
+        };
+
+        IdsEntity parameters = new IdsEntity();
+        parameters.setIdOne(minNodeId);
+        parameters.setIdTwo(maxNodeId);
+        parameters.setIds(qnameIds);
+        parameters.setOrdered(ordered);
+        parameters.setMaxResults(maxResults);
+        template.select(SELECT_NODES_WITH_ASPECT_IDS_LIMITED, parameters, resultHandler);
+    }
+
     @Override
     protected Long insertNodeAssoc(Long sourceNodeId, Long targetNodeId, Long assocTypeQNameId, int assocIndex)
     {

repository/src/main/java/org/alfresco/repo/domain/permissions/FixedAclUpdater.java

@@ -85,8 +85,11 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
 
     public static final String FIXED_ACL_ASYNC_REQUIRED_KEY = "FIXED_ACL_ASYNC_REQUIRED";
     public static final String FIXED_ACL_ASYNC_CALL_KEY = "FIXED_ACL_ASYNC_CALL";
+
     protected static final QName LOCK_Q_NAME = QName.createQName(NamespaceService.SYSTEM_MODEL_1_0_URI, "FixedAclUpdater");
 
+    private static final int DEFAULT_MAX_ITEMS = Integer.MAX_VALUE;
+
     /** A set of listeners to receive callback events whenever permissions are updated by this class. */
     private static Set<FixedAclUpdaterListener> listeners = Sets.newConcurrentHashSet();
 
@@ -101,6 +104,8 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
     private int maxItemBatchSize = 100;
     private int numThreads = 4;
     private boolean forceSharedACL = false;
+    private int maxItems = DEFAULT_MAX_ITEMS;
+    private boolean orderNodes = true;
 
     private ClassPolicyDelegate<OnInheritPermissionsDisabled> onInheritPermissionsDisabledDelegate;
     private PolicyComponent policyComponent;
@@ -147,12 +152,22 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
         this.forceSharedACL = forceSharedACL;
     }
 
+    public void setOrderNodes(boolean orderNodes)
+    {
+        this.orderNodes = orderNodes;
+    }
+
     public void setLockTimeToLive(long lockTimeToLive)
     {
         this.lockTimeToLive = lockTimeToLive;
         this.lockRefreshTime = lockTimeToLive / 2;
     }
 
+    public void setMaxItems(int maxItems)
+    {
+        this.maxItems = maxItems > 0 ? maxItems : DEFAULT_MAX_ITEMS;
+    }
+
     public void setPolicyComponent(PolicyComponent policyComponent)
     {
         this.policyComponent = policyComponent;
@@ -209,7 +224,7 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
                         public List<NodeRef> execute() throws Throwable
                         {
                             getNodesCallback.init();
-                            nodeDAO.getNodesWithAspects(aspects, getNodesCallback.getMinNodeId(), null, true, getNodesCallback);
+                            nodeDAO.getNodesWithAspects(aspects, getNodesCallback.getMinNodeId(), null, orderNodes, maxItemBatchSize, getNodesCallback);
                             getNodesCallback.done();
 
                             return getNodesCallback.getNodes();
@@ -231,6 +246,12 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
                             return countNodesCallback.getCount();
                         }
                     }, false, true);
+
+            if (count > maxItems)
+            {
+                log.info("Total nodes with pending acl: " + count + " Limiting work to " + maxItems);
+                return maxItems;
+            }
             return count;
         }
     }
@@ -238,6 +259,9 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
     private class AclWorkProvider implements BatchProcessWorkProvider<NodeRef>
     {
         private GetNodesWithAspects getNodesWithAspects;
+        private long estimatedUpdatedItems;
+        private long execTime;
+        private long execBatches;
 
         AclWorkProvider()
         {
@@ -259,8 +283,37 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
         @Override
         public Collection<NodeRef> getNextWork()
         {
-            return getNodesWithAspects.getNodesWithAspects();
+            if(estimatedUpdatedItems >= maxItems)
+            {
+                log.info("Reached max items to process. Nodes Processed: " + estimatedUpdatedItems + "/" + maxItems);
+                return Collections.emptyList();
+            }
+
+            long initTime = System.currentTimeMillis();
+            Collection<NodeRef> batchNodes = getNodesWithAspects.getNodesWithAspects();
+            long endTime = System.currentTimeMillis();
+
+            if (log.isDebugEnabled())
+            {
+                log.debug("Query for batch executed in " + (endTime-initTime) + " ms");
+            }
+
+            if (!batchNodes.isEmpty())
+            {
+                // Increment estimatedUpdatedItems with the expected number of nodes to process
+                estimatedUpdatedItems += batchNodes.size();
+                execTime+=endTime-initTime;
+                execBatches++;
+            }
+
+            return batchNodes;
         }
+
+        public double getAverageQueryExecutionTime()
+        {
+            return execBatches > 0 ? execTime/execBatches : 0;
+        }
+
     }
 
     protected class AclWorker implements BatchProcessor.BatchProcessWorker<NodeRef>
@@ -451,6 +504,7 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
 
         try
         {
+            log.info("Running FixedAclUpdater. Max Items: " + maxItems + ", Impose order: " + orderNodes);
             lockToken = jobLockService.getLock(LOCK_Q_NAME, lockTimeToLive, 0, 1);
             jobLockService.refreshLock(lockToken, LOCK_Q_NAME, lockRefreshTime, jobLockRefreshCallback);
 
@@ -460,6 +514,7 @@ public class FixedAclUpdater extends TransactionListenerAdapter implements Appli
                     transactionService.getRetryingTransactionHelper(), provider, numThreads, maxItemBatchSize, applicationContext,
                     log, 100);
             int count = bp.process(worker, true);
+            log.info("FixedAclUpdater updated " + count + ". Average query time " + provider.getAverageQueryExecutionTime() + " ms");
             return count;
         }
         catch (LockAcquisitionException e)

repository/src/main/resources/alfresco/ibatis/org.alfresco.repo.domain.dialect.Dialect/node-common-SqlMap.xml

@@ -782,6 +782,25 @@
         <if test="ordered == true">order by node.id ASC</if>
     </select>
 
+    <select id="select_NodesWithAspectIds_Limited" parameterType="Ids" resultMap="result_NodeRef" >
+        select
+            node.id             as id,
+            store.protocol      as protocol,
+            store.identifier    as identifier,
+            node.uuid           as uuid
+        from
+            alf_node_aspects na
+            join alf_node node on (na.node_id = node.id)
+            left join alf_store store on (store.id = node.store_id)
+        where
+            <![CDATA[na.node_id >= #{idOne}]]>
+            <if test="idTwo != null"><![CDATA[and na.node_id < #{idTwo}]]></if>
+            and na.qname_id in
+                <foreach item="item" index="i" collection="ids" open="(" separator="," close=")">#{item}</foreach>
+        <if test="ordered == true">order by node.id ASC</if>
+        <if test="maxResults != null"><![CDATA[limit #{maxResults}]]></if>
+    </select>
+
     <!-- Common results for result_NodeAssoc -->
     <sql id="select_NodeAssoc_Results">
         select

repository/src/main/resources/alfresco/public-services-security-context.xml

@@ -121,6 +121,8 @@
         <property name="maxItemBatchSize" value="${system.fixedACLsUpdater.maxItemBatchSize}"/>
         <property name="numThreads" value="${system.fixedACLsUpdater.numThreads}"/>
         <property name="forceSharedACL" value="${system.fixedACLsUpdater.forceSharedACL}"/>
+        <property name="maxItems" value="${system.fixedACLsUpdater.maxItems}"/>
+        <property name="orderNodes" value="${system.fixedACLsUpdater.orderNodes}"/>
         <property name="lockTimeToLive" value="${system.fixedACLsUpdater.lockTTL}"/>
         <property name="policyComponent" ref="policyComponent"/>
         <property name="policyIgnoreUtil" ref="policyIgnoreUtil"/>

repository/src/main/resources/alfresco/repository.properties

@@ -1104,7 +1104,11 @@ system.fixedACLsUpdater.numThreads=4
 # fixedACLsUpdater - Force shared ACL to propagate through children even if there is an unexpected ACL
 system.fixedACLsUpdater.forceSharedACL=false
 # fixedACLsUpdater cron expression - fire at midnight every day
-system.fixedACLsUpdater.cronExpression=0 0 0 * * ? 
+system.fixedACLsUpdater.cronExpression=0 0 0 * * ?
+# fixedACLsUpdater - maximum number of pending ACLs to process overall
+system.fixedACLsUpdater.maxItems=-1
+# fixedACLsUpdater - Impose the order by in the query. If false, it may not process all the results but should do the queries faster
+system.fixedACLsUpdater.orderNodes=true
 
 cmis.disable.hidden.leading.period.files=false
 

repository/src/test/java/org/alfresco/repo/domain/permissions/FixedAclUpdaterTest.java

@@ -104,9 +104,9 @@ public class FixedAclUpdaterTest
     private ContentService contentService;
     private AuthorityService authorityService;
     private static final long MAX_TRANSACTION_TIME_DEFAULT = 10;
+    private static final int LARGE_TRANSACTION_TIME = 86_400_000;
     private static final int[] filesPerLevelMoreFolders = { 5, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 };
     private static final int[] filesPerLevelMoreFiles = { 5, 100 };
-    private long maxTransactionTime;
     private static HashMap<Integer, Class<?>> errors;
     private static String TEST_GROUP_NAME = "FixedACLUpdaterTest";
     private static String TEST_GROUP_NAME_FULL = PermissionService.GROUP_PREFIX + TEST_GROUP_NAME;
@@ -134,8 +134,11 @@ public class FixedAclUpdaterTest
         AuthenticationUtil.setFullyAuthenticatedUser(AuthenticationUtil.getSystemUserName());
 
         homeFolderNodeRef = repository.getCompanyHome();
-        maxTransactionTime = MAX_TRANSACTION_TIME_DEFAULT;
+        setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, MAX_TRANSACTION_TIME_DEFAULT);
-        setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, maxTransactionTime);
+
+        fixedAclUpdater.setForceSharedACL(false);
+        fixedAclUpdater.setMaxItems(-1);
+        fixedAclUpdater.setOrderNodes(true);
     }
 
     @After
@@ -155,8 +158,7 @@ public class FixedAclUpdaterTest
 
         try
         {
-            maxTransactionTime = 86400000;
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, LARGE_TRANSACTION_TIME);
-            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, maxTransactionTime);
             setPermissionsOnTree(folderRef, false, false);
             aclComparator.compareACLs();
 
@@ -164,6 +166,7 @@ public class FixedAclUpdaterTest
         }
         finally
         {
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, MAX_TRANSACTION_TIME_DEFAULT);
             deleteNodes(folderRef);
         }
     }
@@ -344,8 +347,7 @@ public class FixedAclUpdaterTest
 
         try
         {
-            maxTransactionTime = 86400000;
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, LARGE_TRANSACTION_TIME);
-            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, maxTransactionTime);
 
             // Set permissions on target folder
             txnHelper.doInTransaction((RetryingTransactionCallback<Void>) () -> {
@@ -386,6 +388,7 @@ public class FixedAclUpdaterTest
         }
         finally
         {
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, MAX_TRANSACTION_TIME_DEFAULT);
             deleteNodes(originalRef);
             deleteNodes(targetRefBase);
         }
@@ -1438,6 +1441,79 @@ public class FixedAclUpdaterTest
         }
     }
 
+    /*
+     * Test with maxItems limit
+     */
+    @Test
+    @RetryAtMost(3)
+    public void testWithLimits()
+    {
+        NodeRef folderRef = createFolderHierarchyInRootForFileTests("testWithLimitsFolder");
+
+        try
+        {
+            int maxItems = 200;
+            setPermissionsOnTree(folderRef, true, true);
+
+            // Get the current amount of pending ACls
+            int initialPendingAcls = getNodesCountWithPendingFixedAclAspect();
+
+            // We need at least maxItems+1 pending ACLs
+            while (initialPendingAcls <= maxItems && initialPendingAcls > 0)
+            {
+                // Trigger the job a single round each time to create new pendings until we have enough
+                triggerFixedACLJob(false,true,maxItems,1);
+                initialPendingAcls = getNodesCountWithPendingFixedAclAspect();
+            }
+
+            assertTrue("We don't have enough pending acls to test", initialPendingAcls > 0);
+
+            // Increase transaction time to not create new pending ACLs
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, LARGE_TRANSACTION_TIME);
+
+            // Trigger job in single round without timeout
+            triggerFixedACLJob(false,true,maxItems,1);
+
+            int finalPendingAcls = getNodesCountWithPendingFixedAclAspect();
+
+            assertTrue("Processed ACLs should not have exceeded 200", (initialPendingAcls - finalPendingAcls) <= maxItems);
+        }
+        finally
+        {
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, MAX_TRANSACTION_TIME_DEFAULT);
+            deleteNodes(folderRef);
+        }
+    }
+
+    /*
+     * Test without imposing the order by
+     */
+    @Test
+    @RetryAtMost(3)
+    public void testUnordered()
+    {
+        NodeRef folderRef = createFolderHierarchyInRootForFileTests("testWithLimitsFolder");
+
+        try
+        {
+            setPermissionsOnTree(folderRef, true, true);
+
+            int initialPendingAcls = getNodesCountWithPendingFixedAclAspect();
+            assertTrue("We don't have enough pending acls to test", initialPendingAcls > 0);
+
+            triggerFixedACLJob(false,true,-1,30);
+
+            int finalPendingAcls = getNodesCountWithPendingFixedAclAspect();
+
+            assertEquals("Not all ACls were processed",0, finalPendingAcls);
+        }
+        finally
+        {
+            setFixedAclMaxTransactionTime(permissionsDaoComponent, homeFolderNodeRef, MAX_TRANSACTION_TIME_DEFAULT);
+            deleteNodes(folderRef);
+        }
+    }
+
     private Long getChild(Long parentId)
     {
         List<FileInfo> children = fileFolderService.list(nodeDAO.getNodePair(parentId).getSecond());
@@ -1601,13 +1677,18 @@ public class FixedAclUpdaterTest
 
     private void triggerFixedACLJob()
     {
-        triggerFixedACLJob(false);
+        // Trigger job 30 times max to process all nodes
+        triggerFixedACLJob(false, true, -1, 30);
     }
 
     private void triggerFixedACLJob(boolean forceSharedACL)
+    {
+        triggerFixedACLJob(forceSharedACL, true, -1, 30);
+    }
+
+    private void triggerFixedACLJob(boolean forceSharedACL, boolean orderNodes, int maxItems, int rounds)
     {
         LOG.debug("Fixing ACL");
-        final int rounds = 30;
         final int enoughZeros = 3;
 
         int numberOfConsecutiveZeros = 0;
@@ -1615,6 +1696,8 @@ public class FixedAclUpdaterTest
         {
             int count = txnHelper.doInTransaction(() -> {
                 fixedAclUpdater.setForceSharedACL(forceSharedACL);
+                fixedAclUpdater.setMaxItems(maxItems);
+                fixedAclUpdater.setOrderNodes(orderNodes);
                 return fixedAclUpdater.execute();
             }, false, true);
             numberOfConsecutiveZeros = count == 0 ? numberOfConsecutiveZeros + 1 : 0;