docs: @GuardedBy( "lock" )
      protected ChangeSet persistChanges(Iterable<NodeKey> changedNodesInOrder) {
          
          // Compute the save meta-info ...
          WorkspaceCache persistedCache = workspaceCache();
          ExecutionContext context = context();
          String userId = context.getSecurityContext().getUserName();
          Map<String, String> userData = context.getData();
          DateTime timestamp = context.getValueFactories().getDateFactory().create();
          String workspaceName = persistedCache.getWorkspaceName();
          String repositoryKey = persistedCache.getRepositoryKey();
          RecordingChanges changes = new RecordingChanges(context.getId(), context.getProcessId(), repositoryKey, workspaceName,
                                                          repositoryEnvironment.journalId());
  
          // Get the documentStore ...
          DocumentStore documentStore = persistedCache.documentStore();
          DocumentTranslator translator = persistedCache.translator();
  
          PathCache sessionPaths = new PathCache(this);
          PathCache workspacePaths = new PathCache(persistedCache);
  
          Set<NodeKey> removedNodes = null;
          Set<NodeKey> removedUnorderedCollections = null;
          Set<BinaryKey> unusedBinaryKeys = new HashSet<>();
          Set<BinaryKey> usedBinaryKeys = new HashSet<>();
          Set<NodeKey> renamedExternalNodes = new HashSet<>();
          Map<NodeKey, Map<BucketId, Set<NodeKey>>> unorderedCollectionBucketRemovals = null;
          
          NodeTypes nodeTypes = nodeTypes();
          for (NodeKey key : changedNodesInOrder) {
              SessionNode node = changedNodes.get(key);
              String keyStr = key.toString();
              boolean isExternal = !node.getKey().getSourceKey().equalsIgnoreCase(workspaceCache().getRootKey().getSourceKey());
              if (node == REMOVED) {
                  // We need to read some information from the node before we remove it ...
                  CachedNode persisted = persistedCache.getNode(key);
                  if (persisted != null) {
                      // This was a persistent node, so we have to generate an event and deal with the remove ...
                      if (removedNodes == null) {
                          removedNodes = new HashSet<>();
                      }
                      Name primaryType = persisted.getPrimaryType(this);
                      Set<Name> mixinTypes = persisted.getMixinTypes(this);
                      boolean isUnorderedCollection = nodeTypes != null && nodeTypes.isUnorderedCollection(primaryType, mixinTypes);
                      if (isUnorderedCollection && removedUnorderedCollections == null) {
                          removedUnorderedCollections = new HashSet<>();
                      }
                     
                      Path path = workspacePaths.getPath(persisted);
                      NodeKey parentKey = persisted.getParentKey(persistedCache);
                      CachedNode parent = getNode(parentKey);
                      Name parentPrimaryType;
                      Set<Name> parentMixinTypes;
                      if (parent != null) {
                          // the parent is loaded in this session
                          parentPrimaryType = parent.getPrimaryType(this);
                          parentMixinTypes = parent.getMixinTypes(this);
                      } else {
                          // the parent has been removed
                          parent = persistedCache.getNode(parentKey);
                          parentPrimaryType = parent.getPrimaryType(persistedCache);
                          parentMixinTypes = parent.getMixinTypes(persistedCache);
                      }
                      changes.nodeRemoved(key, parentKey, path, primaryType, mixinTypes, parentPrimaryType, parentMixinTypes);
                      removedNodes.add(key);
                      if (isUnorderedCollection) {
                          removedUnorderedCollections.add(key);
                      }
  
                      // if there were any referrer changes for the removed nodes, we need to process them
                      ReferrerChanges referrerChanges = referrerChangesForRemovedNodes.get(key);
                      if (referrerChanges != null) {
                          EditableDocument doc = documentStore.edit(keyStr, false);
                          if (doc != null) translator.changeReferrers(doc, referrerChanges);
                      }
  
                      // if the node had any binary properties, make sure we decrement the ref count of each
                      for (Iterator<Property> propertyIterator = persisted.getProperties(persistedCache); propertyIterator.hasNext();) {
                          Property property = propertyIterator.next();
                          if (property.isBinary()) {
                              Object value = property.isMultiple() ? Arrays.asList(property.getValuesAsArray()) : property.getFirstValue();
                              translator.decrementBinaryReferenceCount(value, unusedBinaryKeys, null);
                          }
                      }
  
                      // Note 1: Do not actually remove the document from the documentStore yet; see below (note 2)
                  }
                  // Otherwise, the removed node was created in the session (but not ever persisted),
                  // so we don't have to do anything ...
              } else {
                  // Get the primary and mixin type names; even though we're passing in the session, the two properties
                  // should be there and shouldn't require a looking in the cache...
                  Name primaryType = node.getPrimaryType(this);
                  Set<Name> mixinTypes = node.getMixinTypes(this);
                  boolean isUnorderedCollection = nodeTypes != null && nodeTypes.isUnorderedCollection(primaryType, mixinTypes);
                 
                  CachedNode persisted = null;
                  Path newPath = null;
                  NodeKey newParent = node.newParent();
                  EditableDocument doc = null;
                  ChangedAdditionalParents additionalParents = node.additionalParents();
  
                  if (node.isNew()) {
                      doc = Schematic.newDocument();
                      translator.setKey(doc, key);
                      translator.setParents(doc, newParent, null, additionalParents);
                      translator.addInternalProperties(doc, node.getAddedInternalProperties());
                      
                      SessionNode mutableParent = mutable(newParent);
                      Name parentPrimaryType = mutableParent.getPrimaryType(this);
                      Set<Name> parentMixinTypes = mutableParent.getMixinTypes(this);
                      boolean parentAllowsSNS = nodeTypes != null && nodeTypes.allowsNameSiblings(parentPrimaryType, parentMixinTypes);
                      if (!parentAllowsSNS) {
                          // if the parent of the *new* node doesn't allow SNS, we can optimize the path lookup by
                          // looking directly at the parent's appended nodes because:
                          // a) "this" is a new node
                          // b) the parent must be already in this cache 
                          // c) no SNS are allowed meaning that we don't care about the already existent child references  
                          // and determining the path in a regular fashion *can* very expensive because it requires loading all of the parent's child references
                          MutableChildReferences appended = mutableParent.appended(false);
                          Path parentPath = sessionPaths.getPath(mutableParent);
                          ChildReference ref = appended.getChild(key);
                          if (ref == null) {
                              // the child is new, but doesn't appear in the 'appended' list, so it must've been reordered...
                              ref = mutableParent.changedChildren().inserted(key);
                          }
                          assert ref != null;
                          newPath = pathFactory().create(parentPath, ref.getName());
                          sessionPaths.put(key, newPath);
                      } else {
                          // do a regular lookup of the path, which *will load* the parent's child references
                          newPath = sessionPaths.getPath(node);
                      }                    
                      // Create an event ...
                      changes.nodeCreated(key, newParent, newPath, primaryType, mixinTypes, node.changedProperties());
                  } else {
                      doc = documentStore.edit(keyStr, true);
                      if (doc == null) {
                          if (isExternal && renamedExternalNodes.contains(key)) {
                              // this is a renamed external node which has been processed in the parent, so we can skip it
                              continue;
                          }
                          // Could not find the entry in the documentStore, which means it was deleted by someone else
                          // just moments before we got our transaction to save ...
                          throw new DocumentNotFoundException(keyStr);
                      }
                      // process any internal properties first
                      translator.addInternalProperties(doc, node.getAddedInternalProperties());
                      translator.removeInteralProperties(doc, node.getRemovedInternalProperties());
                      
                      // only after we're certain the document exists in the store can we safely compute this path
                      newPath = sessionPaths.getPath(node);
                      if (newParent != null) {
                          persisted = persistedCache.getNode(key);
                          // The node has moved (either within the same parent or to another parent) ...
                          Path oldPath = workspacePaths.getPath(persisted);
                          NodeKey oldParentKey = persisted.getParentKey(persistedCache);
                          if (!oldParentKey.equals(newParent) || (additionalParents != null && !additionalParents.isEmpty())) {
                              translator.setParents(doc, node.newParent(), oldParentKey, additionalParents);
                          }
                          // We only want to fire the event if the node we're working with is in the same workspace as the current
                          // workspace. The node will be in a different workspace when it is linked or un-linked
                          // (e.g. shareable node or jcr:system).
                          String workspaceKey = node.getKey().getWorkspaceKey();
                          boolean isSameWorkspace = persistedCache.getWorkspaceKey().equalsIgnoreCase(workspaceKey);
                          if (isSameWorkspace) {
                              changes.nodeMoved(key, primaryType, mixinTypes, newParent, oldParentKey, newPath, oldPath);
                          }
                      } else if (additionalParents != null) {
                          // The node in another workspace has been linked to this workspace ...
                          translator.setParents(doc, null, null, additionalParents);
                      }
  
                      // Deal with mixin changes here (since for new nodes they're put into the properties) ...
                      MixinChanges mixinChanges = node.mixinChanges(false);
                      if (mixinChanges != null && !mixinChanges.isEmpty()) {
                          Property oldProperty = translator.getProperty(doc, JcrLexicon.MIXIN_TYPES);
                          translator.addPropertyValues(doc, JcrLexicon.MIXIN_TYPES, true, mixinChanges.getAdded(),
                                                       unusedBinaryKeys, usedBinaryKeys);
                          translator.removePropertyValues(doc, JcrLexicon.MIXIN_TYPES, mixinChanges.getRemoved(), unusedBinaryKeys,
                                                          usedBinaryKeys);
                          // the property was changed ...
                          Property newProperty = translator.getProperty(doc, JcrLexicon.MIXIN_TYPES);
                          if (oldProperty == null) {
                              changes.propertyAdded(key, primaryType, mixinTypes, newPath, newProperty);
                          } else if (newProperty == null) {
                              changes.propertyRemoved(key, primaryType, mixinTypes, newPath, oldProperty);
                          } else {
                              changes.propertyChanged(key, primaryType, mixinTypes, newPath, newProperty, oldProperty);
                          }
                      }
                  }
  
                  LockChange lockChange = node.getLockChange();
                  if (lockChange != null) {
                      switch (lockChange) {
                          case LOCK_FOR_SESSION:
                          case LOCK_FOR_NON_SESSION:
                              // check if another session has already locked the document
                              if (translator.isLocked(doc)) {
                                  throw new LockFailureException(key);
                              }
                              break;
                          case UNLOCK:
                              break;
                      }
                  }
  
                  // As we go through the removed and changed properties, we want to keep track of whether there are any
                  // effective modifications to the persisted properties.
                  boolean hasPropertyChanges = false;
  
                  // Save the removed properties ...
                  Set<Name> removedProperties = node.removedProperties();
                  if (!removedProperties.isEmpty()) {
                      assert !node.isNew();
                      if (persisted == null) {
                          persisted = persistedCache.getNode(key);
                      }
                      for (Name name : removedProperties) {
                          Property oldProperty = translator.removeProperty(doc, name, unusedBinaryKeys, usedBinaryKeys);
                          if (oldProperty != null) {
                              // the property was removed ...
                              changes.propertyRemoved(key, primaryType, mixinTypes, newPath, oldProperty);
                              // and we know that there are modifications to the properties ...
                              hasPropertyChanges = true;
                          }
                      }
                  }
  
                  // Save the changes to the properties
                  if (!node.changedProperties().isEmpty()) {
                      if (!node.isNew() && persisted == null) {
                          persisted = persistedCache.getNode(key);
                      }
                      for (Map.Entry<Name, Property> propEntry : node.changedProperties().entrySet()) {
                          Name name = propEntry.getKey();
                          Property prop = propEntry.getValue();
                          // Get the old property ...
                          Property oldProperty = persisted != null ? persisted.getProperty(name, persistedCache) : null;
                          translator.setProperty(doc, prop, unusedBinaryKeys, usedBinaryKeys);
                          if (oldProperty == null) {
                              // the property was created ...
                              changes.propertyAdded(key, primaryType, mixinTypes, newPath, prop);
                              // and we know that there are modifications to the properties ...
                              hasPropertyChanges = true;
                          } else if (!oldProperty.equals(prop)) {
                              // the property was changed and is actually different than the persisted property ...
                              changes.propertyChanged(key, primaryType, mixinTypes, newPath, prop, oldProperty);
                              hasPropertyChanges = true;
                          }
                      }
                  }
  
                  // Save the change to the child references. Note that we only need to generate events for renames;
                  // moves (to the same or another parent), removes, and inserts are all recorded as changes in the
                  // child node, and events are generated handled when we process
                  // the child node.
                  ChangedChildren changedChildren = node.changedChildren();
                  MutableChildReferences appended = node.appended(false);
                  if ((changedChildren == null || changedChildren.isEmpty()) && (appended != null && !appended.isEmpty())) {
                      // Just appended children ...  
                      if (!isUnorderedCollection) {
                          translator.changeChildren(doc, changedChildren, appended);
                      } else {
                          translator.addChildrenToBuckets(doc, appended);
                      }
                  } else if (changedChildren != null && !changedChildren.isEmpty()) {
                      ChildReferences childReferences = node.getChildReferences(this);
                      if (!changedChildren.getRemovals().isEmpty() && !isUnorderedCollection) {
                          // This node is not being removed (or added), but it has removals, and we have to calculate the paths
                          // of the removed nodes before we actually change the child references of this node.
                          for (NodeKey removed : changedChildren.getRemovals()) {
                              CachedNode persistedChild = persistedCache.getNode(removed);
                              if (persistedChild != null) {
                                  NodeKey childKey = persistedChild.getKey();
                                  if (appended != null && appended.hasChild(childKey)) {
                                      // the same node has been both removed and appended => reordered at the end
                                      ChildReference appendedChildRef = childReferences.getChild(childKey);
                                      Path newNodePath = pathFactory().create(newPath, appendedChildRef.getSegment());
                                      Path oldNodePath = workspacePaths.getPath(persistedChild);
                                      Map<NodeKey, Map<Path, Path>> pathChangesForSNS = computePathChangesForSNS(workspacePaths, 
                                                                                                                newPath, childReferences,
                                                                                                                appendedChildRef,
                                                                                                                newNodePath, oldNodePath);
                                      changes.nodeReordered(childKey, persistedChild.getPrimaryType(this),
                                                            persistedChild.getMixinTypes(this), node.getKey(), newNodePath,
                                                            oldNodePath, null, pathChangesForSNS);
                                  }
                              }
                          }
                      }
  
                      // Now change the children ...
                      if (!isUnorderedCollection) {
                          // this is a regular node
                          translator.changeChildren(doc, changedChildren, appended);
                      } else {
                          // there are both added & removed children for this collection
                          if (appended != null && !appended.isEmpty()) {
                              // process additions first
                              translator.addChildrenToBuckets(doc, appended);
                          }
                          if (changedChildren.removalCount() > 0){
                              // when removing nodes from unordered collections, a 2 step process is required
                              // 1. collect all the nodes which will be removed from each bucket, without persisting the actual bucket
                              // changes in ISPN. This is because if bucket changes are persisted here, we may have to process a deleted
                              // child later on while not having any more information
                              // 2. after all the changed nodes have been processed, make the actual changes to the bucket documents
                              // see below
                              Map<BucketId, Set<NodeKey>> removalsPerBucket = translator.preRemoveChildrenFromBuckets(workspaceCache(),
                                                                                                                      doc,
                                                                                                                      changedChildren.getRemovals());
                              if (!removalsPerBucket.isEmpty()) {
                                  if (unorderedCollectionBucketRemovals == null) {
                                      unorderedCollectionBucketRemovals = new LinkedHashMap<>();
                                  }
                                  unorderedCollectionBucketRemovals.put(key, removalsPerBucket);
                              }
                          }
                      }
  
                      // Generate events for renames, as this is only captured in the parent node ...
                      Map<NodeKey, Name> newNames = changedChildren.getNewNames();
                      if (!newNames.isEmpty()) {
                          for (Map.Entry<NodeKey, Name> renameEntry : newNames.entrySet()) {
                              NodeKey renamedKey = renameEntry.getKey();
                              CachedNode oldRenamedNode = persistedCache.getNode(renamedKey);
                              if (oldRenamedNode == null) {
                                  // The node was created in this session, so we can ignore this ...
                                  continue;
                              }
                              Path renamedFromPath = workspacePaths.getPath(oldRenamedNode);
                              Path renamedToPath = pathFactory().create(renamedFromPath.getParent(), renameEntry.getValue());
                              changes.nodeRenamed(renamedKey, renamedToPath, renamedFromPath.getLastSegment(), primaryType,
                                                  mixinTypes);
                              if (isExternal) {
                                  renamedExternalNodes.add(renamedKey);
                              }
                          }
                      }
  
                      // generate reordering events for nodes which have not been reordered to the end
                      Map<NodeKey, SessionNode.Insertions> insertionsByBeforeKey = changedChildren.getInsertionsByBeforeKey();
                      for (SessionNode.Insertions insertion : insertionsByBeforeKey.values()) {
                          for (ChildReference insertedRef : insertion.inserted()) {
                              CachedNode insertedNodePersistent = persistedCache.getNode(insertedRef);
                              // if the node is new and reordered at the same time (most likely due to either a version restore
                              // or explicit reordering of transient nodes) there is no "old path"
                              CachedNode insertedNode = getNode(insertedRef.getKey());
                              if (insertedNodePersistent != null) {
                                  // the reordered node exists (is not new) so to get its new path (after the reordering) we need
                                  // to look at the session's view of the parent
                                  ChildReference nodeNewRef = childReferences.getChild(insertedRef.getKey());
                                  Path nodeNewPath = pathFactory().create(newPath, nodeNewRef.getSegment());
                                  Path nodeOldPath = workspacePaths.getPath(insertedNodePersistent);
                                  Path insertedBeforePath = null;
                                  CachedNode insertedBeforeNode = persistedCache.getNode(insertion.insertedBefore());
                                  if (insertedBeforeNode != null) {
                                      insertedBeforePath = workspacePaths.getPath(insertedBeforeNode);
                                      boolean isSnsReordering = nodeOldPath.getLastSegment().getName()
                                                                           .equals(insertedBeforePath.getLastSegment().getName());
                                      if (isSnsReordering) {
                                          nodeNewPath = insertedBeforePath;
                                      }
                                  }
                                  Map<NodeKey, Map<Path, Path>> snsPathChangesByNodeKey = computePathChangesForSNS(
                                          workspacePaths, newPath, childReferences, insertedRef, nodeNewPath, nodeOldPath);
                                  changes.nodeReordered(insertedRef.getKey(), insertedNode.getPrimaryType(this),
                                                        insertedNode.getMixinTypes(this), node.getKey(), nodeNewPath, nodeOldPath,
                                                        insertedBeforePath, snsPathChangesByNodeKey);
  
                              } else {
                                  Path nodeNewPath = sessionPaths.getPath(insertedNode);
                                  CachedNode insertedBeforeNode = getNode(insertion.insertedBefore().getKey());
                                  Path insertedBeforePath = sessionPaths.getPath(insertedBeforeNode);
                                  changes.nodeReordered(insertedRef.getKey(), insertedNode.getPrimaryType(this),
                                                        insertedNode.getMixinTypes(this), node.getKey(), nodeNewPath, null,
                                                        insertedBeforePath);
                              }
                          }
                      }
                  }
  
                  ReferrerChanges referrerChanges = node.getReferrerChanges();
                  boolean nodeChanged = false;
                  if (referrerChanges != null && !referrerChanges.isEmpty()) {
                      translator.changeReferrers(doc, referrerChanges);
                      changes.nodeChanged(key, newPath, primaryType, mixinTypes);
                      nodeChanged = true;
                  }
  
                  // write the federated segments
                  for (Map.Entry<String, String> federatedSegment : node.getAddedFederatedSegments().entrySet()) {
                      String externalNodeKey = federatedSegment.getKey();
                      String childName = federatedSegment.getValue();
                      translator.addFederatedSegment(doc, externalNodeKey, childName);
                      if (!nodeChanged) {
                          changes.nodeChanged(key, newPath, primaryType, mixinTypes);
                          nodeChanged = true;
                      }
                  }
                  Set<String> removedFederatedSegments = node.getRemovedFederatedSegments();
                  if (!removedFederatedSegments.isEmpty()) {
                      translator.removeFederatedSegments(doc, node.getRemovedFederatedSegments());
                      if (!nodeChanged) {
                          changes.nodeChanged(key, newPath, primaryType, mixinTypes);
                          nodeChanged = true;
                      }
                  }
  
                  // write additional node "metadata", meaning various flags which have internal meaning
                  boolean excludedFromSearch = node.isExcludedFromSearch(this);
                  if (excludedFromSearch && translator.isQueryable(doc)) {
                      translator.setQueryable(doc, false);
                  }
                  if (!excludedFromSearch && !translator.isQueryable(doc)) {
                      translator.setQueryable(doc, true);
                  }
  
                  if (node.isNew()) {
                      // We need to create the schematic entry for the new node ...
                      if (documentStore.storeIfAbsent(keyStr, doc) != null) {
                          if (replacedNodes != null && replacedNodes.contains(key)) {
                              // Then a node is being removed and recreated with the same key ...
                              documentStore.localStore().put(keyStr, doc);
                          } else if (removedNodes != null && removedNodes.contains(key)) {
                              // Then a node is being removed and recreated with the same key ...
                              documentStore.localStore().put(keyStr, doc);
                              removedNodes.remove(key);
                          } else {
                              // We couldn't create the entry because one already existed ...
                              throw new DocumentAlreadyExistsException(keyStr);
                          }
                      }
                  } else {
                      boolean externalNodeChanged = isExternal
                                                    && (hasPropertyChanges || node.hasNonPropertyChanges() || node.changedChildren()
                                                                                                                  .renameCount() > 0);
  
                      // writable connectors *may* change their data in-place, so the update operation needs to be called only
                      // after the index changes have finished.
                      if (externalNodeChanged) {
                          // in the case of external nodes, only if there are changes should the update be called
                          documentStore.updateDocument(keyStr, doc, node);
                      }
                  }
  
                  // The above code doesn't properly generate events for newly linked or unlinked nodes (e.g., shareable nodes
                  // in JCR), because NODE_ADDED or NODE_REMOVED events are generated based upon the creation or removal of the
                  // child nodes, whereas linking and unlinking nodes don't result in creation/removal of nodes. Instead,
                  // the linked/unlinked node is modified with the addition/removal of additional parents.
                  //
                  // NOTE that this happens somewhat rarely (as linked/shared nodes are used far less frequently) ...
                  //
                  if (additionalParents != null) {
                      // Generate NODE_ADDED events for each of the newly-added parents ...
                      for (NodeKey parentKey : additionalParents.getAdditions()) {
                          // Find the mutable parent node (if it exists) ...
                          SessionNode parent = this.changedNodes.get(parentKey);
                          if (parent != null) {
                              // Then the parent was changed in this session, so find the one-and-only child reference ...
                              ChildReference ref = parent.getChildReferences(this).getChild(key);
                              Path parentPath = sessionPaths.getPath(parent);
                              Path childPath = pathFactory().create(parentPath, ref.getSegment());
                              changes.nodeCreated(key, parentKey, childPath, primaryType, mixinTypes, null);
                          }
                      }
                      // Generate NODE_REMOVED events for each of the newly-removed parents ...
                      for (NodeKey parentKey : additionalParents.getRemovals()) {
                          // We need to read some information from the parent node before it was changed ...
                          CachedNode persistedParent = persistedCache.getNode(parentKey);
                          if (persistedParent != null) {
                              // Find the path to the removed child ...
                              ChildReference ref = persistedParent.getChildReferences(this).getChild(key);
                              if (ref != null) {
                                  Path parentPath = workspacePaths.getPath(persistedParent);
                                  Path childPath = pathFactory().create(parentPath, ref.getSegment());
                                  Name parentPrimaryType = persistedParent.getPrimaryType(persistedCache);
                                  Set<Name> parentMixinTypes = persistedParent.getMixinTypes(persistedCache);
                                  changes.nodeRemoved(key, parentKey, childPath, primaryType, mixinTypes,
                                                      parentPrimaryType, parentMixinTypes);
                              }
                          }
                      }
                  }
              }
          }
          
          // persist any bucket document changes resulted from removing children from unordered collections
          if (unorderedCollectionBucketRemovals != null) {
              for (Map.Entry<NodeKey, Map<BucketId, Set<NodeKey>>> entry : unorderedCollectionBucketRemovals.entrySet()) {
                  translator.persistBucketRemovalChanges(entry.getKey(), entry.getValue());                
              }
          }
  
          // Remove all the buckets from all the unordered collections which have been removed...
          // This will only remove the buckets, not the original nodes which are processed below
          if (removedUnorderedCollections != null) {
              for (NodeKey removedUnorderedCollectionKey : removedUnorderedCollections) {
                  translator.removeAllBucketsFromUnorderedCollection(removedUnorderedCollectionKey);
              }
          }
  
          if (removedNodes != null) {
              assert !removedNodes.isEmpty();
              // we need to collect the referrers at the end only, so that other potential changes in references have been computed
              Map<NodeKey, Set<NodeKey>> referrersByRemovedNodes = new HashMap<>();
           
              for (NodeKey removedKey : removedNodes) {
                  // we need the current document from the documentStore, because may differs from what's persisted (i.e. the latest
                  // persisted information
                  SchematicEntry entry = documentStore.get(removedKey.toString());
                  if (entry != null) {
                      // The entry hasn't yet been removed by another (concurrent) session ...
                      Document doc = entry.content();
                      Set<NodeKey> strongReferrers = translator.getReferrers(doc, ReferenceType.STRONG);
                      strongReferrers.removeAll(removedNodes);
                      if (!strongReferrers.isEmpty()) {
                          referrersByRemovedNodes.put(removedKey, strongReferrers);
                      }
                  }
              }
  
              if (!referrersByRemovedNodes.isEmpty()) {
                  Map<String, Set<String>> referrerPathsByRemovedNode = new HashMap<>();
                  for (Map.Entry<NodeKey, Set<NodeKey>> entry : referrersByRemovedNodes.entrySet()) {
                      String removedNodePath = workspacePaths.getPath(persistedCache.getNode(entry.getKey())).toString();
                      Set<NodeKey> referrerKeys = entry.getValue();
                      Set<String> referrerPaths = new HashSet<>(referrerKeys.size());
                      for (NodeKey referrerKey : referrerKeys) {
                          referrerPaths.add(sessionPaths.getPath(getNode(referrerKey)).toString());
                      }
                      referrerPathsByRemovedNode.put(removedNodePath, referrerPaths);
                  }
                  throw new ReferentialIntegrityException(referrerPathsByRemovedNode);
              }
  
              // Now remove all of the nodes from the documentStore.
              // Note 2: we do this last because the children are removed from their parent before the removal is handled above
              // (see Node 1), meaning getting the path and other information for removed nodes never would work properly.
              for (NodeKey removedKey : removedNodes) {
                  documentStore.remove(removedKey.toString());
              }
          }
  
          if (!unusedBinaryKeys.isEmpty()) {
              // There are some binary values that are no longer referenced ...
              for (BinaryKey key : unusedBinaryKeys) {
                  changes.binaryValueNoLongerUsed(key);
              }
          }
  
          if (!usedBinaryKeys.isEmpty()) {
              // There are some binary values which need to be marked as used ...
              for (BinaryKey key : usedBinaryKeys) {
                  changes.binaryValueUsed(key);
              }
          }
  
          changes.setChangedNodes(changedNodes.keySet()); // don't need to make a copy
          changes.freeze(userId, userData, timestamp);
          return changes;
      }