[#171] Skip CSV snapshot rows without a uid and delete the snapshots a failed sync copied

OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java

@@ -463,10 +463,13 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha
 
         final String[] header = getHeader();
         File syncOrigin = null;
+        // the snapshots this call copies from the live file; deleted again if the sync fails
+        final List<File> copies = new ArrayList<File>(2);
         boolean changesProcessed = false;
 
         // start with a write-lock
         final ReentrantReadWriteLock rwLock = fileNameToLockMap.get(csvFilePath);
+        boolean downgraded = false;
         rwLock.writeLock().lock();
         try {
             if (token != null && token.getValue() != null) {
@@ -482,17 +485,26 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha
                 } catch (IOException e) {
                     throw new ConnectorException("Unable to copy CSV file for sync operation", e);
                 }
+                copies.add(syncOrigin);
                 token = new SyncToken(timestamp);
             }
             scrubSyncFiles();
 
             // downgrade to a read-lock
             rwLock.readLock().lock();
+            downgraded = true;
         } finally {
+            if (!downgraded) {
+                // the read section below deletes the copies only once it is entered
+                for (File copy : copies) {
+                    deleteRewriteFile(copy);
+                }
+            }
             rwLock.writeLock().unlock();
         }
 
         // we now have a read-lock unless the above block threw an Exception
+        boolean completed = false;
         try {
             final MessageDigest digest = newFingerprintDigest();
             // uid of every current row -> fingerprint of the first origin row with that uid, null while there is none
@@ -562,10 +574,19 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha
                 boolean reportDeletes = true;
                 while ((entry = reader.read(originHeader, processors)) != null) {
                     final String uid = (String) entry.get(config.getHeaderUid());
+                    if (StringUtil.isBlank(uid)) {
+                        // a row without a uid was never reported, so there is no object to delete
+                        log.warn("Skipping line {0} of {1}: it has no {2} value",
+                                String.valueOf(reader.getLineNumber()), syncOrigin.toString(), config.getHeaderUid());
+                        continue;
+                    }
                     if (!index.containsKey(uid)) {
                         if (reportDeletes) {
                             SyncDelta delta = generateSyncDelta(SyncDeltaType.DELETE, newConnectorObject(entry), token);
-                            if (handler.handle(delta)) {
+                            final boolean more = handler.handle(delta);
+                            // the handler may have stored the token of this delta, so its snapshot must stay
+                            copies.clear();
+                            if (more) {
                                 changesProcessed = true;
                             } else {
                                 // keep reading: the next pass needs the fingerprints of the remaining origin rows
@@ -612,7 +633,10 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha
                         continue;
                     }
                     SyncDelta delta = generateSyncDelta(type, newConnectorObject(entry), token);
-                    if (!handler.handle(delta)) {
+                    final boolean more = handler.handle(delta);
+                    // the handler may have stored the token of this delta, so its snapshot must stay
+                    copies.clear();
+                    if (!more) {
                         break;
                     }
                     changesProcessed = true;
@@ -643,15 +667,24 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha
                     } catch (IOException e) {
                         throw new ConnectorException("Unable to copy CSV file for sync operation", e);
                     }
+                    copies.add(syncOrigin);
                     token = new SyncToken(timestamp);
                 }
             }
 
             if (handler instanceof SyncTokenResultsHandler) {
                 ((SyncTokenResultsHandler) handler).handleResult(token);
             }
+            completed = true;
         } finally {
             rwLock.readLock().unlock();
+            if (!completed) {
+                // a kept copy could be handed out by getLatestSyncToken,
+                // and it would stop a retry from copying the unchanged live file again
+                for (File copy : copies) {
+                    deleteSnapshot(rwLock, copy);
+                }
+            }
         }
     }
 
@@ -999,6 +1032,18 @@ private void scrubSyncFiles() {
         }
     }
 
+    /**
+     * Deletes a snapshot under the write lock, so that no sync in this JVM is reading it meanwhile.
+     */
+    private void deleteSnapshot(ReentrantReadWriteLock rwLock, File snapshot) {
+        rwLock.writeLock().lock();
+        try {
+            deleteRewriteFile(snapshot);
+        } finally {
+            rwLock.writeLock().unlock();
+        }
+    }
+
     private void compareHeaders(String[] header2) {
         List<String> hdr1 = Arrays.asList(getHeader());
         List<String> hdr2 = Arrays.asList(header2);

OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java

@@ -16,18 +16,29 @@
 package org.forgerock.openicf.csvfile;
 
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.expectThrows;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.PrintStream;
+import java.io.UncheckedIOException;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystems;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.nio.file.attribute.PosixFilePermission;
+import java.nio.file.attribute.PosixFilePermissions;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import org.identityconnectors.common.security.GuardedString;
@@ -43,6 +54,7 @@
 import org.identityconnectors.framework.common.objects.SyncToken;
 import org.identityconnectors.framework.spi.SyncTokenResultsHandler;
 import org.supercsv.exception.SuperCsvException;
+import org.testng.SkipException;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.DataProvider;
@@ -58,6 +70,10 @@ public class SyncOpDiffTest {
 
     private static final SyncToken SNAPSHOT_TOKEN = new SyncToken(1300000000000L);
 
+    // the second row of each uid differs from the first one, so even a fresh copy of this file reports two UPDATEs
+    private static final String[] DUPLICATE_UIDS =
+            { HEADER, "x,Jane,Doe,pw1", "x,Jane,Changed,pw1", "y,John,Roe,pw2", "y,John,Other,pw2" };
+
     private Path dir;
 
     private File csvFile;
@@ -176,6 +192,163 @@ public void rowWithoutUidFailsSyncBeforeAnyDelta(String row) throws Exception {
         assertEquals(e.getMessage(), "Line 3 of " + csvFile + " has no uid value");
     }
 
+    @Test(dataProvider = "rowsWithoutUid")
+    public void snapshotRowWithoutUidIsSkipped(String row) throws Exception {
+        givenSnapshot(HEADER, "u1,Ann,Roe,pw0", row, "u3,Bob,Poe,pw3");
+        givenLive(HEADER, "u3,Bob,Poe,pw3", "u2,Jane,Doe,pw1");
+
+        assertEquals(sync(SNAPSHOT_TOKEN), Arrays.asList("DELETE u1 [Ann, Roe, pw0]", "CREATE u2 [Jane, Doe, pw1]"));
+    }
+
+    @Test
+    public void skippedSnapshotRowIsLogged() throws Exception {
+        givenSnapshot(HEADER, "u1,Jane,Doe,pw1", ",John,Roe,pw2");
+        givenLive(HEADER, "u1,Jane,Doe,pw1");
+
+        String output = captureStdOut(() -> sync(SNAPSHOT_TOKEN));
+
+        assertTrue(output.contains("Skipping line 3 of "
+                + dir.resolve(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue()) + ": it has no uid value"), output);
+    }
+
+    @Test
+    public void failedSyncDeletesTheSnapshotItCopied() throws Exception {
+        givenLive(HEADER, "u1,Jane,Doe,pw1", ",John,Roe,pw2");
+
+        expectThrows(IllegalArgumentException.class, () -> sync(null));
+
+        assertEquals(snapshots(), Collections.emptyList());
+    }
+
+    @Test
+    public void failedSyncKeepsTheSnapshotOfItsToken() throws Exception {
+        givenSnapshot(HEADER, "u1,Jane,Doe,pw1");
+        givenLive(HEADER, "u1,Jane,Doe,pw1", ",John,Roe,pw2");
+
+        expectThrows(IllegalArgumentException.class, () -> sync(SNAPSHOT_TOKEN));
+
+        assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue()));
+    }
+
+    @Test
+    public void successfulSyncKeepsTheSnapshotItCopied() throws Exception {
+        givenLive(HEADER, "u1,Jane,Doe,pw1");
+
+        sync(null);
+
+        assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + resultToken.getValue()));
+    }
+
+    @Test
+    public void syncWhoseTokenIsNotDeliveredDeletesTheSnapshotItCopied() throws Exception {
+        givenLive(HEADER, "u1,Jane,Doe,pw1");
+
+        expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, () -> { }));
+
+        assertEquals(snapshots(), Collections.emptyList());
+    }
+
+    @Test
+    public void syncWhoseTokenIsNotDeliveredDeletesTheSnapshotItCopiedAfterTheDiff() throws Exception {
+        givenSnapshot(HEADER, "u1,Jane,Doe,pw1");
+        givenLive(HEADER, "u1,Jane,Doe,pw1", "u2,John,Roe,pw2");
+
+        expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(SNAPSHOT_TOKEN, () -> { }));
+
+        assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue()));
+    }
+
+    @Test
+    public void failedSyncKeepsTheSnapshotOfADeltaItHandled() throws Exception {
+        givenLive(DUPLICATE_UIDS);
+
+        List<SyncToken> handled = syncFailingAfter(1);
+
+        assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + handled.get(0).getValue()));
+    }
+
+    @Test
+    public void failedSyncKeepsTheSnapshotOfADeleteItHandled() throws Exception {
+        givenLive(HEADER, "u1,Jane,Doe,pw1", "u2,John,Roe,pw2");
+        connector();
+        // relies on the header cached at init: the live rows are read with it, so every live uid is a password,
+        // while a fresh copy is read with its own header and reports two DELETEs
+        givenLive("password,lastName,uid,firstName", "pw1,Doe,u1,Jane", "pw2,Roe,u2,John");
+
+        List<SyncToken> handled = syncFailingAfter(1);
+
+        assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + handled.get(0).getValue()));
+    }
+
+    @Test
+    public void syncWhoseFirstDeltaFailsDeletesTheSnapshotItCopied() throws Exception {
+        givenLive(DUPLICATE_UIDS);
+
+        syncFailingAfter(0);
+
+        assertEquals(snapshots(), Collections.emptyList());
+    }
+
+    @Test
+    public void snapshotThatCannotBeDeletedIsLogged() throws Exception {
+        if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) {
+            throw new SkipException("A read-only directory needs POSIX permissions");
+        }
+        final Set<PosixFilePermission> mode = Files.getPosixFilePermissions(dir);
+        final Set<PosixFilePermission> readOnly = PosixFilePermissions.fromString("r-xr-xr-x");
+        setDirMode(readOnly);
+        final boolean writable = Files.isWritable(dir);
+        setDirMode(mode);
+        if (writable) {
+            throw new SkipException("A read-only directory does not stop this user from deleting files");
+        }
+        givenLive(HEADER, "u1,Jane,Doe,pw1");
+        final File copy = new File(dir.toFile(), csvFile.getName() + "." + csvFile.lastModified());
+
+        final String output;
+        try {
+            output = captureStdOut(() -> expectThrows(IllegalStateException.class,
+                    () -> syncWithFailingHandleResult(null, () -> setDirMode(readOnly))));
+        } finally {
+            setDirMode(mode);
+        }
+
+        assertTrue(output.contains("Could not delete " + copy), output);
+    }
+
+    @Test
+    public void syncThatCannotListItsSnapshotsDeletesTheSnapshotItCopied() throws Exception {
+        if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) {
+            throw new SkipException("An unlistable directory needs POSIX permissions");
+        }
+        final Set<PosixFilePermission> mode = Files.getPosixFilePermissions(dir);
+        // files can still be created in the directory, but not listed
+        final Set<PosixFilePermission> writeOnly = PosixFilePermissions.fromString("-wx------");
+        setDirMode(writeOnly);
+        final boolean listable = dir.toFile().list() != null;
+        setDirMode(mode);
+        if (listable) {
+            throw new SkipException("A directory without read permission does not stop this user from listing it");
+        }
+        givenLive(HEADER, "u1,Jane,Doe,pw1");
+        connector();
+        // only creating or deleting an entry in the directory moves its mtime off this value
+        final FileTime untouched = FileTime.fromMillis(0);
+        Files.setLastModifiedTime(dir, untouched);
+
+        setDirMode(writeOnly);
+        try {
+            // the copy is made, then scrubSyncFiles fails while the sync still holds the write lock;
+            // how it fails is not the point here
+            expectThrows(RuntimeException.class, () -> sync(null));
+        } finally {
+            setDirMode(mode);
+        }
+
+        assertNotEquals(Files.getLastModifiedTime(dir), untouched, "The sync failed before it copied the CSV file");
+        assertEquals(snapshots(), Collections.emptyList());
+    }
+
     @Test
     public void rowWithWrongColumnCountFailsSyncBeforeAnyDelta() throws Exception {
         givenSnapshot(HEADER, "u1,Jane,Doe,pw1");
@@ -298,6 +471,64 @@ private void givenLive(String... lines) throws IOException {
         Files.write(csvFile.toPath(), Arrays.asList(lines), StandardCharsets.UTF_8);
     }
 
+    // the connector logs through StdOutLogger, which reads System.out on every call
+    private static String captureStdOut(Runnable action) {
+        final PrintStream original = System.out;
+        final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(buffer, true));
+        try {
+            action.run();
+        } finally {
+            System.setOut(original);
+        }
+        return buffer.toString();
+    }
+
+    private List<String> snapshots() throws IOException {
+        try (Stream<Path> paths = Files.list(dir)) {
+            return paths.map(path -> path.getFileName().toString())
+                    .filter(name -> name.startsWith(csvFile.getName() + "."))
+                    .sorted()
+                    .collect(Collectors.toList());
+        }
+    }
+
+    private void syncWithFailingHandleResult(final SyncToken token, final Runnable beforeFailing) {
+        connector().sync(ObjectClass.ACCOUNT, token, new SyncTokenResultsHandler() {
+            @Override
+            public boolean handle(SyncDelta delta) {
+                return true;
+            }
+
+            @Override
+            public void handleResult(SyncToken syncToken) {
+                beforeFailing.run();
+                throw new IllegalStateException("token not stored");
+            }
+        }, null);
+    }
+
+    // syncs from no token, handles the given number of deltas and fails on the next one
+    private List<SyncToken> syncFailingAfter(final int count) {
+        final List<SyncToken> handled = new ArrayList<>();
+        expectThrows(IllegalStateException.class, () -> connector().sync(ObjectClass.ACCOUNT, null, delta -> {
+            if (handled.size() == count) {
+                throw new IllegalStateException("delta not stored");
+            }
+            handled.add(delta.getToken());
+            return true;
+        }, null));
+        return handled;
+    }
+
+    private void setDirMode(Set<PosixFilePermission> mode) {
+        try {
+            Files.setPosixFilePermissions(dir, mode);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
     private CSVFileConnector connector() {
         if (connector == null) {
             CSVFileConfiguration config = new CSVFileConfiguration();