From c3d7af643411ca8f1b40dd06a95e15e4d9cc9c06 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 10:33:27 +0300 Subject: [PATCH 01/14] [#174] Match CSV sync snapshot names exactly --- .../openicf/csvfile/CSVFileConnector.java | 49 ++++++------ .../openicf/csvfile/SyncOpDiffTest.java | 75 +++++++++++++++++++ 2 files changed, 99 insertions(+), 25 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index c82ab6dd..dd8278d2 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -47,7 +47,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; @@ -716,23 +715,12 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha public SyncToken getLatestSyncToken(ObjectClass objectClass) { isAccount(objectClass); - File dataDir = config.getCsvFile().getParentFile(); - Map files = new TreeMap(); - final ReadLock lock = fileNameToLockMap.get(csvFilePath).readLock(); lock.lock(); try { - Pattern filePattern = Pattern.compile("(" + config.getCsvFile().getName() + ")(\\.[0-9]{13})$"); - for (String filename : dataDir.list()) { - Matcher matcher = filePattern.matcher(filename); - if (matcher.find()) { - Long timestamp = Long.valueOf(matcher.group(2).substring(1)); - files.put(timestamp, filename); - } - } - if (files.size() > 0) { - List keys = new LinkedList(files.keySet()); - return new SyncToken(keys.get(keys.size() - 1)); + final TreeMap snapshots = listSnapshots(); + if (!snapshots.isEmpty()) { + return new SyncToken(snapshots.lastKey()); } return new SyncToken(new Date().getTime()); } finally { @@ -1035,18 +1023,29 @@ private static void updateLength(MessageDigest digest, int length) { digest.update((byte) length); } - private void scrubSyncFiles() { - File dataDir = config.getCsvFile().getParentFile(); - Map files = new TreeMap(); - - Pattern filePattern = Pattern.compile("(" + config.getCsvFile().getName() + ")(\\.[0-9]{13})$"); - for (String filename : dataDir.list()) { - Matcher matcher = filePattern.matcher(filename); - if (matcher.find()) { - Long timestamp = Long.valueOf(matcher.group(2).substring(1)); - files.put(timestamp, filename); + /** + * Lists the sync snapshots of the CSV file, {@code .<13-digit mtime>}, oldest first. The name is matched + * exactly, so the snapshots of another CSV file in the same directory are never listed. + */ + private TreeMap listSnapshots() { + final File dataDir = config.getCsvFile().getParentFile(); + final String[] names = dataDir.list(); + if (names == null) { + throw new ConnectorIOException("Unable to list the files of " + dataDir); + } + final Pattern snapshotName = Pattern.compile(Pattern.quote(config.getCsvFile().getName()) + "\\.([0-9]{13})"); + final TreeMap snapshots = new TreeMap(); + for (String name : names) { + final Matcher matcher = snapshotName.matcher(name); + if (matcher.matches()) { + snapshots.put(Long.valueOf(matcher.group(1)), new File(dataDir, name)); } } + return snapshots; + } + + private void scrubSyncFiles() { + final Map files = listSnapshots(); while (files.size() > config.getSyncFileRetentionCount()) { Long key = (Long) files.keySet().toArray()[0]; files.remove(key); diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index 8ec68f5d..bd60fc20 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -594,6 +594,77 @@ public void snapshotHeaderMismatchFailsBeforeTheCurrentRowsAreRead() throws Exce sync(SNAPSHOT_TOKEN); } + @DataProvider + public Object[][] otherFileNames() { + return new Object[][] { + // the old pattern was not anchored at the start + { "old_data.csv" }, + // and did not quote the dot + { "dataXcsv" } + }; + } + + @Test(dataProvider = "otherFileNames") + public void latestSyncTokenIgnoresSnapshotsOfOtherFiles(String otherName) throws Exception { + givenSnapshot(HEADER, "u1,Jane,Doe,pw1"); + givenLive(HEADER, "u1,Jane,Doe,pw1"); + givenFile(otherName + ".1600000000000", HEADER); + + assertEquals(connector().getLatestSyncToken(ObjectClass.ACCOUNT), SNAPSHOT_TOKEN); + } + + @DataProvider + public Object[][] regexMetacharacterNames() { + return new Object[][] { + // never matched its own snapshots + { "data(1).csv" }, + // made the pattern invalid + { "data[1.csv" } + }; + } + + @Test(dataProvider = "regexMetacharacterNames") + public void latestSyncTokenFindsTheSnapshotsOfAFileWithRegexMetacharactersInItsName(String name) + throws Exception { + csvFile = dir.resolve(name).toFile(); + givenSnapshot(HEADER, "u1,Jane,Doe,pw1"); + givenLive(HEADER, "u1,Jane,Doe,pw1"); + + assertEquals(connector().getLatestSyncToken(ObjectClass.ACCOUNT), SNAPSHOT_TOKEN); + } + + @Test + public void syncOfAFileWhoseNameIsAnInvalidRegexReportsItsChanges() throws Exception { + csvFile = dir.resolve("data[1.csv").toFile(); + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + } + + @Test + public void latestSyncTokenOfAnUnlistableDirectoryFails() throws Exception { + if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + throw new SkipException("An unlistable directory needs POSIX permissions"); + } + final Set mode = Files.getPosixFilePermissions(dir); + givenLive(HEADER, "u1,Jane,Doe,pw1"); + connector(); + + setDirMode(PosixFilePermissions.fromString("-wx------")); + try { + if (dir.toFile().list() != null) { + throw new SkipException("A directory without read permission does not stop this user from listing it"); + } + // the old code failed with a NullPointerException + ConnectorIOException e = expectThrows(ConnectorIOException.class, + () -> connector.getLatestSyncToken(ObjectClass.ACCOUNT)); + assertEquals(e.getMessage(), "Unable to list the files of " + dir.toFile()); + } finally { + setDirMode(mode); + } + } + @Test public void missingCsvFileFailsSync() throws Exception { givenSnapshot(HEADER, "u1,Jane,Doe,pw1"); @@ -673,6 +744,10 @@ private void givenLive(String... lines) throws IOException { Files.write(csvFile.toPath(), Arrays.asList(lines), StandardCharsets.UTF_8); } + private Path givenFile(String name, String... lines) throws IOException { + return Files.write(dir.resolve(name), 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; From f578943ce2e8e3142ad429fc0dc0a5e36d8169e7 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 10:36:44 +0300 Subject: [PATCH 02/14] [#174] Reuse an existing CSV snapshot of the same mtime and advance the token to it --- .../openicf/csvfile/CSVFileConnector.java | 38 ++++++-- .../openicf/csvfile/SyncOpDiffTest.java | 87 ++++++++++++++++++- 2 files changed, 117 insertions(+), 8 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index dd8278d2..97f3af48 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -484,11 +484,13 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha syncOrigin = new File(config.getCsvFile().getParentFile(), config.getCsvFile().getName() + "." + timestamp); try { - Files.copy(config.getCsvFile().toPath(), syncOrigin.toPath()); + // a snapshot of this mtime taken by an earlier sync is the baseline, and not this call's to delete + if (takeSnapshot(syncOrigin)) { + copies.add(syncOrigin); + } } catch (IOException e) { throw new ConnectorException("Unable to copy CSV file for sync operation", e); } - copies.add(syncOrigin); token = new SyncToken(timestamp); } scrubSyncFiles(); @@ -682,7 +684,8 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha copies.add(syncOrigin); token = new SyncToken(nextTimestamp); } catch (FileAlreadyExistsException e) { - // the snapshot of this mtime is kept, and so is the token + // taken by an earlier sync of the same changes, such as another consumer's; not added to copies + token = new SyncToken(nextTimestamp); } catch (IOException e) { throw new ConnectorException("Unable to copy CSV file for sync operation", e); } @@ -695,8 +698,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } 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 + // a kept copy could be handed out by getLatestSyncToken for (File copy : copies) { deleteSnapshot(rwLock, copy); } @@ -1315,6 +1317,32 @@ File copyCsvFile() throws IOException { } } + /** + * Takes a snapshot of the CSV file: a complete copy, renamed into place, so that no snapshot is ever seen + * half-written. Makes nothing if a snapshot of that name exists, as it holds the same mtime of the CSV file. + * + * @return whether this call made the snapshot + */ + private boolean takeSnapshot(File snapshot) throws IOException { + if (snapshot.exists()) { + return false; + } + final File copy = copyCsvFile(); + boolean moved = false; + try { + Files.move(copy.toPath(), snapshot.toPath()); + moved = true; + } catch (FileAlreadyExistsException e) { + // taken meanwhile by another process + } finally { + if (!moved) { + // once moved, the name is free again, and createRewriteFile may hand it to a writer + deleteRewriteFile(copy); + } + } + return moved; + } + /** * Keeps the cached row count in step with a create or a delete. A count * that has not been taken yet (-1) stays so: the next paged search takes it. diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index bd60fc20..ec26ac9a 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -523,20 +523,101 @@ File copyCsvFile() throws IOException { } @Test - public void existingSnapshotOfTheCopiedMtimeKeepsTheToken() throws Exception { + public void existingSnapshotOfTheCopiedMtimeAdvancesTheToken() throws Exception { givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); givenLive(HEADER, "k,Ann,Poe,pw3"); final Path existing = dir.resolve(csvFile.getName() + "." + csvFile.lastModified()); - final List existingLines = Arrays.asList(HEADER, "x,Ann,Poe,pw3"); + // the live rows, written differently, so that a replaced snapshot shows + final List existingLines = Arrays.asList(HEADER, "\"k\",\"Ann\",\"Poe\",\"pw3\""); Files.write(existing, existingLines, StandardCharsets.UTF_8); assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); - assertEquals(resultToken.getValue(), SNAPSHOT_TOKEN.getValue()); + assertEquals(resultToken.getValue(), csvFile.lastModified()); assertEquals(Files.readAllLines(existing, StandardCharsets.UTF_8), existingLines); assertEquals(snapshots(), Arrays.asList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue(), existing.getFileName().toString())); } + @Test + public void failedSyncKeepsTheExistingSnapshotOfTheCopiedMtime() throws Exception { + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + final Path existing = givenFile(csvFile.getName() + "." + csvFile.lastModified(), HEADER, "k,Ann,Poe,pw3"); + + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(SNAPSHOT_TOKEN, () -> { })); + + assertTrue(Files.exists(existing)); + } + + @Test + public void secondConsumerOfTheSameChangesGetsTheNewTokenToo() throws Exception { + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + final SyncToken first = resultToken; + // another consumer stored the same token; the old code gave it that token back, so it got this delta forever + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + + assertEquals(resultToken, first); + assertEquals(sync(resultToken), Collections.emptyList()); + } + + @Test + public void secondSyncWithoutATokenOfAnUnchangedFileReusesItsSnapshot() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + final List copies = connectorRecordingCopies(false); + + assertEquals(sync(null), Collections.emptyList()); + final SyncToken first = resultToken; + // the old code failed here: its copy found the snapshot of the first sync in the way + assertEquals(sync(null), Collections.emptyList()); + + assertEquals(resultToken, first); + assertEquals(first.getValue(), csvFile.lastModified()); + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + first.getValue())); + // the snapshot and the copy for the diff, then only the copy for the diff + assertEquals(copies.size(), 3); + } + + @Test + public void failedSyncKeepsTheSnapshotItReused() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + sync(null); + final List taken = snapshots(); + + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, () -> { })); + + assertEquals(snapshots(), taken); + } + + @Test + public void snapshotTakenMeanwhileByAnotherProcessIsKept() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + final Path snapshot = dir.resolve(csvFile.getName() + "." + csvFile.lastModified()); + // the live row, written differently, so that a replaced snapshot shows + final List theirs = Arrays.asList(HEADER, "\"u1\",\"Jane\",\"Doe\",\"pw1\""); + final List copies = new ArrayList<>(); + connector(new CSVFileConnector() { + @Override + File copyCsvFile() throws IOException { + final File copy = super.copyCsvFile(); + if (copies.isEmpty()) { + // another process takes the snapshot of this mtime while the sync copies the CSV file for it + Files.write(snapshot, theirs, StandardCharsets.UTF_8); + } + copies.add(copy); + return copy; + } + }); + + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, () -> { })); + + assertEquals(Files.readAllLines(snapshot, StandardCharsets.UTF_8), theirs); + // the copy that lost the race is deleted + assertEquals(snapshots(), Collections.singletonList(snapshot.getFileName().toString())); + } + @Test public void snapshotThatCannotBeMovedInFailsSync() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { From 5b9139ba014d837266a7c593d2812e2d46abd2eb Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 10:39:47 +0300 Subject: [PATCH 03/14] [#174] Delete the CSV sync snapshots beyond syncFileRetentionCount --- .../openicf/csvfile/CSVFileConnector.java | 22 +++-- .../openicf/csvfile/SyncOpDiffTest.java | 88 +++++++++++++++++++ 2 files changed, 104 insertions(+), 6 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index 97f3af48..dc06bf92 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -493,7 +493,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } token = new SyncToken(timestamp); } - scrubSyncFiles(); + scrubSyncFiles(syncOrigin); // downgrade to a read-lock rwLock.readLock().lock(); @@ -1046,11 +1046,21 @@ private TreeMap listSnapshots() { return snapshots; } - private void scrubSyncFiles() { - final Map files = listSnapshots(); - while (files.size() > config.getSyncFileRetentionCount()) { - Long key = (Long) files.keySet().toArray()[0]; - files.remove(key); + /** + * Deletes the snapshots older than the {@code syncFileRetentionCount} newest ones, except the snapshot this sync + * reads. Runs under the write lock, so no sync in this JVM reads a snapshot while it is deleted; one in another + * JVM may. A snapshot that cannot be deleted, such as one still open on Windows, is logged and kept. + */ + private void scrubSyncFiles(File syncOrigin) { + final TreeMap snapshots = listSnapshots(); + int surplus = snapshots.size() - config.getSyncFileRetentionCount(); + for (File snapshot : snapshots.values()) { + if (surplus-- <= 0) { + break; + } + if (!snapshot.equals(syncOrigin)) { + deleteRewriteFile(snapshot); + } } } diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index ec26ac9a..fb8563c6 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -92,12 +92,15 @@ public class SyncOpDiffTest { private SyncToken resultToken; + private Integer retentionCount; + @BeforeMethod public void before() throws IOException { dir = Files.createTempDirectory("csv-sync"); csvFile = dir.resolve("data.csv").toFile(); deltas = null; resultToken = null; + retentionCount = null; } @AfterMethod(alwaysRun = true) @@ -618,6 +621,88 @@ File copyCsvFile() throws IOException { assertEquals(snapshots(), Collections.singletonList(snapshot.getFileName().toString())); } + @Test + public void syncDeletesTheSnapshotsBeyondTheRetentionCount() throws Exception { + retentionCount = 2; + givenFile(csvFile.getName() + ".1100000000000", HEADER, "k,Ann,Poe,pw3"); + givenFile(csvFile.getName() + ".1200000000000", HEADER, "k,Ann,Poe,pw3"); + givenSnapshot(HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.emptyList()); + + assertEquals(snapshots(), Arrays.asList(csvFile.getName() + ".1200000000000", + csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue())); + } + + @Test + public void syncKeepsTheSnapshotItReadsWhenItIsTheOldest() throws Exception { + retentionCount = 1; + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenFile(csvFile.getName() + ".1400000000000", HEADER, "k,Ann,Poe,pw3"); + givenFile(csvFile.getName() + ".1500000000000", HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + + // the newest snapshot and the one this sync read are kept, then the sync adds its own + assertEquals(snapshots(), Arrays.asList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue(), + csvFile.getName() + ".1500000000000", csvFile.getName() + "." + resultToken.getValue())); + } + + @Test + public void syncKeepsTheSnapshotsOfOtherFiles() throws Exception { + retentionCount = 1; + givenSnapshot(HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + // each would be the oldest snapshot under the old pattern, which was not anchored and did not quote the dot + final List others = Arrays.asList("old_data.csv.1100000000000", "dataXcsv.1200000000000"); + for (String other : others) { + givenFile(other, HEADER, "k,Ann,Poe,pw3"); + } + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.emptyList()); + + for (String other : others) { + assertTrue(Files.exists(dir.resolve(other)), other); + } + } + + @Test + public void snapshotThatCannotBeScrubbedIsLogged() throws Exception { + if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + throw new SkipException("A read-only directory needs POSIX permissions"); + } + final Set mode = Files.getPosixFilePermissions(dir); + final Set 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"); + } + retentionCount = 1; + final Path old = givenFile(csvFile.getName() + ".1200000000000", HEADER, "k,Ann,Poe,pw3"); + givenSnapshot(HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + connector(); + + final List failures = new ArrayList<>(); + final String output; + setDirMode(readOnly); + try { + // the scrub only logs; then the copy for the diff fails in the read-only directory + output = captureStdOut(() -> failures.add(expectThrows(ConnectorException.class, + () -> sync(SNAPSHOT_TOKEN)))); + } finally { + setDirMode(mode); + } + + assertTrue(output.contains("Could not delete " + old.toFile()), output); + assertEquals(failures.get(0).getMessage(), "Unable to copy CSV file for sync operation"); + assertTrue(Files.exists(old)); + } + @Test public void snapshotThatCannotBeMovedInFailsSync() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { @@ -938,6 +1023,9 @@ private CSVFileConnector connector(CSVFileConnector instance) { config.setCsvFile(csvFile); config.setHeaderUid("uid"); config.setHeaderPassword("password"); + if (retentionCount != null) { + config.setSyncFileRetentionCount(retentionCount); + } instance.init(config); connector = instance; return connector; From 2c0760aa70adde5d5bad0886c2d86db0fd5b03b6 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 10:42:09 +0300 Subject: [PATCH 04/14] [#174] Fail sync() from a token without a snapshot and take one in getLatestSyncToken --- .../openicf/csvfile/CSVFileConnector.java | 20 +++++++++++--- .../openicf/csvfile/SyncOpDiffTest.java | 26 +++++++++++++++++++ 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index dc06bf92..d71da714 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -42,7 +42,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.Date; import java.util.EnumSet; import java.util.HashMap; import java.util.HashSet; @@ -478,8 +477,13 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha if (token != null && token.getValue() != null) { syncOrigin = new File(config.getCsvFile().getParentFile(), config.getCsvFile().getName() + "." + token.getValue()); - } - if (syncOrigin == null || !syncOrigin.exists()) { + if (!syncOrigin.exists()) { + // a baseline taken now would hide every delete since the token + throw new ConnectorException("Snapshot " + syncOrigin + " of sync token " + token.getValue() + + " does not exist, so the changes since that token cannot be reported;" + + " reset the sync token"); + } + } else { long timestamp = config.getCsvFile().lastModified(); syncOrigin = new File(config.getCsvFile().getParentFile(), config.getCsvFile().getName() + "." + timestamp); @@ -724,7 +728,15 @@ public SyncToken getLatestSyncToken(ObjectClass objectClass) { if (!snapshots.isEmpty()) { return new SyncToken(snapshots.lastKey()); } - return new SyncToken(new Date().getTime()); + // every token handed out has a snapshot, so the first sync from it reports every change made since + final long timestamp = config.getCsvFile().lastModified(); + try { + takeSnapshot(new File(config.getCsvFile().getParentFile(), + config.getCsvFile().getName() + "." + timestamp)); + } catch (IOException e) { + throw new ConnectorException("Unable to copy CSV file for sync operation", e); + } + return new SyncToken(timestamp); } finally { lock.unlock(); } diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index fb8563c6..fe3739c8 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -703,6 +703,32 @@ public void snapshotThatCannotBeScrubbedIsLogged() throws Exception { assertTrue(Files.exists(old)); } + @Test + public void syncFromATokenWhoseSnapshotIsGoneFails() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + final File snapshot = new File(dir.toFile(), csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue()); + + ConnectorException e = expectThrows(ConnectorException.class, () -> sync(SNAPSHOT_TOKEN)); + + assertEquals(e.getMessage(), "Snapshot " + snapshot + " of sync token " + SNAPSHOT_TOKEN.getValue() + + " does not exist, so the changes since that token cannot be reported; reset the sync token"); + assertEquals(deltas, Collections.emptyList()); + assertEquals(snapshots(), Collections.emptyList()); + } + + @Test + public void latestSyncTokenWithoutASnapshotTakesOne() throws Exception { + givenLive(HEADER, "k,Ann,Poe,pw3"); + + final SyncToken token = connector().getLatestSyncToken(ObjectClass.ACCOUNT); + + assertEquals(token.getValue(), csvFile.lastModified()); + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + token.getValue())); + // the first sync from the token reports a change made after the call; the old code lost it + givenLive(HEADER, "k,Ann,Poe,pw3", "c,Jane,Doe,pw1"); + assertEquals(sync(token), Collections.singletonList("CREATE c [Jane, Doe, pw1]")); + } + @Test public void snapshotThatCannotBeMovedInFailsSync() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { From b542f0dca9e27cc4d039ccb6cb9ffaf8f5e3f189 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 11:02:48 +0300 Subject: [PATCH 05/14] [#174] Delete a failed sync's snapshots before it releases the read lock A sync queued on the write lock could take it between the failing sync's read-lock release and its write-locked delete, reuse the failing sync's snapshot and hand out its token, which the delete then left without a snapshot. The read lock keeps other syncs out just as the write lock did. --- .../openicf/csvfile/CSVFileConnector.java | 19 ++------ .../openicf/csvfile/SyncOpDiffTest.java | 44 +++++++++++++++++++ 2 files changed, 48 insertions(+), 15 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index d71da714..0976d1fc 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -700,13 +700,14 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } completed = true; } finally { - rwLock.readLock().unlock(); if (!completed) { - // a kept copy could be handed out by getLatestSyncToken + // a kept copy could be handed out by getLatestSyncToken; deleted before the read lock is released, + // which, like the write lock, keeps every other sync in this JVM out, as one could reuse the copy for (File copy : copies) { - deleteSnapshot(rwLock, copy); + deleteRewriteFile(copy); } } + rwLock.readLock().unlock(); if (next != null) { // not a snapshot, so no other sync reads it deleteRewriteFile(next); @@ -1076,18 +1077,6 @@ private void scrubSyncFiles(File syncOrigin) { } } - /** - * 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 hdr1 = Arrays.asList(getHeader()); List hdr2 = Arrays.asList(header2); diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index fe3739c8..42c1e9a6 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -25,6 +25,7 @@ import java.io.IOException; import java.io.PrintStream; import java.io.UncheckedIOException; +import java.lang.reflect.Field; import java.nio.charset.StandardCharsets; import java.nio.file.AccessDeniedException; import java.nio.file.FileSystems; @@ -38,6 +39,7 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -45,6 +47,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -525,6 +528,29 @@ File copyCsvFile() throws IOException { } } + @Test + public void syncQueuedBehindAFailingSyncGetsATokenWithASnapshot() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + connector(); + // a fair lock hands the write lock to the queued sync before the failing sync can take it back + final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true); + lockMap().put(csvFile.getAbsolutePath(), lock); + final ExecutorService other = Executors.newSingleThreadExecutor(); + final List> queued = new ArrayList<>(); + try { + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, () -> { + queued.add(other.submit(() -> sync(null))); + awaitQueued(lock); + })); + await(queued.get(0)); + + // the old code deleted the snapshot of the failing sync after the queued sync had reused it + assertEquals(sync(resultToken), Collections.emptyList()); + } finally { + other.shutdownNow(); + } + } + @Test public void existingSnapshotOfTheCopiedMtimeAdvancesTheToken() throws Exception { givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); @@ -1012,6 +1038,24 @@ private static T await(Future task) { } } + // waits until another thread queues for the lock + private static void awaitQueued(ReentrantReadWriteLock lock) { + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (!lock.hasQueuedThreads()) { + if (System.nanoTime() > deadline) { + throw new AssertionError("No thread queued for " + lock); + } + Thread.yield(); + } + } + + @SuppressWarnings("unchecked") + private static Map lockMap() throws ReflectiveOperationException { + final Field field = CSVFileConnector.class.getDeclaredField("fileNameToLockMap"); + field.setAccessible(true); + return (Map) field.get(null); + } + private void setDirMode(Set mode) { try { Files.setPosixFilePermissions(dir, mode); From efe60fa9a6fd33e4b0bda56872b48029574096e9 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 12:04:21 +0300 Subject: [PATCH 06/14] [#174] Pin a syncFileRetentionCount of 0 and below A sync keeps only the snapshot it reads, for a token and for its own baseline. Kills an early return for a count below 1 and a scrub that spares only a reused origin. --- .../openicf/csvfile/SyncOpDiffTest.java | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index 42c1e9a6..91a80f75 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -676,6 +676,38 @@ public void syncKeepsTheSnapshotItReadsWhenItIsTheOldest() throws Exception { csvFile.getName() + ".1500000000000", csvFile.getName() + "." + resultToken.getValue())); } + @DataProvider + public Object[][] retentionCountsBelowOne() { + return new Object[][] { { 0 }, { -1 } }; + } + + @Test(dataProvider = "retentionCountsBelowOne") + public void syncFromATokenKeepsOnlyTheSnapshotItReads(int count) throws Exception { + retentionCount = count; + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenFile(csvFile.getName() + ".1400000000000", HEADER, "k,Ann,Poe,pw3"); + givenFile(csvFile.getName() + ".1500000000000", HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + + // then the sync adds its own + assertEquals(snapshots(), Arrays.asList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue(), + csvFile.getName() + "." + resultToken.getValue())); + } + + @Test(dataProvider = "retentionCountsBelowOne") + public void syncWithoutATokenKeepsOnlyItsBaseline(int count) throws Exception { + retentionCount = count; + givenFile(csvFile.getName() + ".1400000000000", HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(null), Collections.emptyList()); + + // with a count of 1 or more the baseline is the newest snapshot, so only a count below 1 reaches it + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + resultToken.getValue())); + } + @Test public void syncKeepsTheSnapshotsOfOtherFiles() throws Exception { retentionCount = 1; From 137a77cc98d546ec3d740423827c1df5c55a7d1c Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 12:05:10 +0300 Subject: [PATCH 07/14] [#174] Describe what syncFileRetentionCount keeps in its help text The scrub runs when a sync starts and never deletes the snapshot that sync reads, so "number of sync history files to retain" undercounts what stays on disk. --- .../org/forgerock/openicf/csvfile/Messages.properties | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/OpenICF-csvfile-connector/src/main/resources/org/forgerock/openicf/csvfile/Messages.properties b/OpenICF-csvfile-connector/src/main/resources/org/forgerock/openicf/csvfile/Messages.properties index 2c8496ff..320282b9 100644 --- a/OpenICF-csvfile-connector/src/main/resources/org/forgerock/openicf/csvfile/Messages.properties +++ b/OpenICF-csvfile-connector/src/main/resources/org/forgerock/openicf/csvfile/Messages.properties @@ -11,6 +11,7 @@ # own identifying information: "Portions copyright [year] [name of copyright owner]". # # Copyright 2015-2016 ForgeRock AS. +# Portions Copyright 2026 3A Systems, LLC. # connector_name.display=CSV File Connector basic.group=Basic configuration properties @@ -27,4 +28,4 @@ csv_field_delimiter.help=Character used to delimit columnar fields csv_lineend_character.display=Newline string csv_lineend_character.help=Character(s) used to terminate a line in the CSV file sync_file_retention_count.display=Historical sync file count -sync_file_retention_count.help=Number of sync history files to retain +sync_file_retention_count.help=Number of the newest sync snapshots of the CSV file that each sync keeps when it starts, besides the snapshot it reads; the sync may then add its own. A count of 0 or below keeps only the snapshot the sync reads From dbf77a381df8f9c4005cb719a65351fcb2488cc0 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 12:07:57 +0300 Subject: [PATCH 08/14] [#174] Build every CSV sync snapshot path from the absolute CSV file A CSV file named without a directory has no parent file, so listing its snapshots failed with a NullPointerException in every sync and in getLatestSyncToken. The listing now uses the absolute directory, and so does the one helper that names a snapshot: a relative origin would never equal its listed file, and the scrub would delete the snapshot the sync is about to read. --- .../openicf/csvfile/CSVFileConnector.java | 24 +++++++----- .../openicf/csvfile/SyncOpDiffTest.java | 38 ++++++++++++++++++- 2 files changed, 52 insertions(+), 10 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index 0976d1fc..91617baa 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -475,8 +475,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha rwLock.writeLock().lock(); try { if (token != null && token.getValue() != null) { - syncOrigin = new File(config.getCsvFile().getParentFile(), - config.getCsvFile().getName() + "." + token.getValue()); + syncOrigin = snapshotFile(token.getValue()); if (!syncOrigin.exists()) { // a baseline taken now would hide every delete since the token throw new ConnectorException("Snapshot " + syncOrigin + " of sync token " + token.getValue() @@ -485,8 +484,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } } else { long timestamp = config.getCsvFile().lastModified(); - syncOrigin = new File(config.getCsvFile().getParentFile(), - config.getCsvFile().getName() + "." + timestamp); + syncOrigin = snapshotFile(timestamp); try { // a snapshot of this mtime taken by an earlier sync is the baseline, and not this call's to delete if (takeSnapshot(syncOrigin)) { @@ -679,8 +677,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } if (changesProcessed) { - syncOrigin = new File(config.getCsvFile().getParentFile(), - config.getCsvFile().getName() + "." + nextTimestamp); + syncOrigin = snapshotFile(nextTimestamp); try { Files.move(next.toPath(), syncOrigin.toPath()); // the name is free again, and createRewriteFile may hand it to a writer @@ -732,8 +729,7 @@ public SyncToken getLatestSyncToken(ObjectClass objectClass) { // every token handed out has a snapshot, so the first sync from it reports every change made since final long timestamp = config.getCsvFile().lastModified(); try { - takeSnapshot(new File(config.getCsvFile().getParentFile(), - config.getCsvFile().getName() + "." + timestamp)); + takeSnapshot(snapshotFile(timestamp)); } catch (IOException e) { throw new ConnectorException("Unable to copy CSV file for sync operation", e); } @@ -1038,12 +1034,22 @@ private static void updateLength(MessageDigest digest, int length) { digest.update((byte) length); } + /** + * Returns the sync snapshot of the CSV file of the given mtime, {@code .}. Its path is absolute, as are + * those {@link #listSnapshots()} lists, so that it equals its listed file even for a CSV file named without a + * directory, which has no parent file. + */ + private File snapshotFile(Object mtime) { + final File csv = config.getCsvFile().getAbsoluteFile(); + return new File(csv.getParentFile(), csv.getName() + "." + mtime); + } + /** * Lists the sync snapshots of the CSV file, {@code .<13-digit mtime>}, oldest first. The name is matched * exactly, so the snapshots of another CSV file in the same directory are never listed. */ private TreeMap listSnapshots() { - final File dataDir = config.getCsvFile().getParentFile(); + final File dataDir = config.getCsvFile().getAbsoluteFile().getParentFile(); final String[] names = dataDir.list(); if (names == null) { throw new ConnectorIOException("Unable to list the files of " + dataDir); diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index 91a80f75..c85fd08c 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -31,6 +31,7 @@ import java.nio.file.FileSystems; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.Paths; import java.nio.file.attribute.FileTime; import java.nio.file.attribute.PosixFilePermission; import java.nio.file.attribute.PosixFilePermissions; @@ -892,6 +893,37 @@ public void syncOfAFileWhoseNameIsAnInvalidRegexReportsItsChanges() throws Excep assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); } + @Test + public void syncOfAFileNamedWithoutADirectoryKeepsTheSnapshotItReads() throws Exception { + // the name resolves against the working directory, but its File has no parent + final Path workingDir = Paths.get("").toAbsolutePath(); + final String name = Files.createTempFile(workingDir, "csv-sync-", ".csv").getFileName().toString(); + try { + csvFile = new File(name); + // the scrub reaches the snapshot the sync reads, so it has to recognise that snapshot in its own listing + retentionCount = 1; + Files.write(workingDir.resolve(name + "." + SNAPSHOT_TOKEN.getValue()), + Arrays.asList(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"), StandardCharsets.UTF_8); + Files.write(workingDir.resolve(name + ".1400000000000"), Arrays.asList(HEADER, "k,Ann,Poe,pw3"), + StandardCharsets.UTF_8); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + // the old code failed with a NullPointerException in the scrub, and so did getLatestSyncToken + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + assertEquals(connector.getLatestSyncToken(ObjectClass.ACCOUNT), resultToken); + + assertEquals(snapshots(workingDir), Arrays.asList(name + "." + SNAPSHOT_TOKEN.getValue(), + name + ".1400000000000", name + "." + resultToken.getValue())); + } finally { + try (Stream paths = Files.list(workingDir)) { + for (Path file : paths.filter(path -> path.getFileName().toString().startsWith(name)) + .collect(Collectors.toList())) { + Files.delete(file); + } + } + } + } + @Test public void latestSyncTokenOfAnUnlistableDirectoryFails() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { @@ -1025,7 +1057,11 @@ private static String captureStdErr(Runnable action) { } private List snapshots() throws IOException { - try (Stream paths = Files.list(dir)) { + return snapshots(dir); + } + + private List snapshots(Path in) throws IOException { + try (Stream paths = Files.list(in)) { return paths.map(path -> path.getFileName().toString()) .filter(name -> name.startsWith(csvFile.getName() + ".")) .sorted() From f268e2b467bb42d44773fb29710287e09ef03597 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 12:13:11 +0300 Subject: [PATCH 09/14] [#174] Advance the CSV sync token to an existing snapshot only if it holds the same bytes Every snapshot is named after the mtime read before its copy, so one a sync in another JVM took may also hold a write made after that mtime was read. Adopting it as the token hid that write from this consumer for good. The end of a sync now adopts an existing snapshot only when it holds the same bytes as the sync's copy; otherwise the token stays, and the next sync reports these changes again. A snapshot that cannot be read is logged and not adopted. --- .../openicf/csvfile/CSVFileConnector.java | 37 +++++++++- .../openicf/csvfile/SyncOpDiffTest.java | 67 +++++++++++++++++-- 2 files changed, 97 insertions(+), 7 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index 91617baa..334fe479 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -26,6 +26,7 @@ import java.io.FileReader; import java.io.FileWriter; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStreamWriter; import java.nio.channels.FileChannel; import java.nio.charset.StandardCharsets; @@ -685,8 +686,13 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha copies.add(syncOrigin); token = new SyncToken(nextTimestamp); } catch (FileAlreadyExistsException e) { - // taken by an earlier sync of the same changes, such as another consumer's; not added to copies - token = new SyncToken(nextTimestamp); + // taken by an earlier sync of the same changes, such as another consumer's; not added to copies. + // Named after the mtime read before its copy, one taken in another JVM may also hold a later + // write, which this sync has not reported: then the token stays, and the next sync reports these + // changes again instead of never reporting that write + if (sameBytes(syncOrigin, next)) { + token = new SyncToken(nextTimestamp); + } } catch (IOException e) { throw new ConnectorException("Unable to copy CSV file for sync operation", e); } @@ -1360,6 +1366,33 @@ private boolean takeSnapshot(File snapshot) throws IOException { return moved; } + /** + * Whether an existing snapshot holds the same bytes as a copy of the CSV file. A snapshot that cannot be read, + * such as one deleted meanwhile in another JVM, does not. + */ + private static boolean sameBytes(File snapshot, File copy) { + if (snapshot.length() != copy.length()) { + return false; + } + try (InputStream in = Files.newInputStream(snapshot.toPath()); + InputStream other = Files.newInputStream(copy.toPath())) { + final byte[] buffer = new byte[8192]; + final byte[] otherBuffer = new byte[buffer.length]; + int length; + do { + length = in.readNBytes(buffer, 0, buffer.length); + if (other.readNBytes(otherBuffer, 0, otherBuffer.length) != length + || !Arrays.equals(buffer, 0, length, otherBuffer, 0, length)) { + return false; + } + } while (length > 0); + return true; + } catch (IOException e) { + log.warn(e, "Could not compare {0} with {1}", snapshot, copy); + return false; + } + } + /** * Keeps the cached row count in step with a create or a delete. A count * that has not been taken yet (-1) stays so: the next paged search takes it. diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index c85fd08c..deac4af0 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -556,18 +556,75 @@ public void syncQueuedBehindAFailingSyncGetsATokenWithASnapshot() throws Excepti public void existingSnapshotOfTheCopiedMtimeAdvancesTheToken() throws Exception { givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); givenLive(HEADER, "k,Ann,Poe,pw3"); - final Path existing = dir.resolve(csvFile.getName() + "." + csvFile.lastModified()); - // the live rows, written differently, so that a replaced snapshot shows - final List existingLines = Arrays.asList(HEADER, "\"k\",\"Ann\",\"Poe\",\"pw3\""); - Files.write(existing, existingLines, StandardCharsets.UTF_8); + // the bytes of the live file, as an earlier sync of the same changes copied them + final Path existing = givenFile(csvFile.getName() + "." + csvFile.lastModified(), HEADER, "k,Ann,Poe,pw3"); assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); assertEquals(resultToken.getValue(), csvFile.lastModified()); - assertEquals(Files.readAllLines(existing, StandardCharsets.UTF_8), existingLines); assertEquals(snapshots(), Arrays.asList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue(), existing.getFileName().toString())); } + @DataProvider + public Object[][] laterWrites() { + return new Object[][] { + { "w,Will,Woe,pw4", "k,Ann,Poe,pw3", "CREATE w [Will, Woe, pw4]" }, + // the same length, and past the first 8 KiB the comparison reads at once + { null, "k,Ann,Poe,pw4", "UPDATE k [Ann, Poe, pw4]" } + }; + } + + @Test(dataProvider = "laterWrites") + public void existingSnapshotOfTheCopiedMtimeWithOtherBytesKeepsTheToken(String added, String k, String delta) + throws Exception { + final List rows = new ArrayList<>(Collections.singletonList(HEADER)); + for (int i = 0; i < 400; i++) { + rows.add(String.format("u%03d,First,Last,pw%03d", i, i)); + } + rows.add("k,Ann,Poe,pw3"); + givenLive(rows.toArray(new String[0])); + rows.add("d,John,Roe,pw2"); + givenSnapshot(rows.toArray(new String[0])); + // another JVM read the same mtime, but copied the file only after a writer had made a change + final List theirs = new ArrayList<>(rows.subList(0, rows.size() - 2)); + theirs.add(k); + if (added != null) { + theirs.add(added); + } + final Path existing = givenFile(csvFile.getName() + "." + csvFile.lastModified(), + theirs.toArray(new String[0])); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.singletonList("DELETE d [John, Roe, pw2]")); + // as the new token, the existing snapshot would hide that change from this consumer for good + assertEquals(resultToken, SNAPSHOT_TOKEN); + assertEquals(Files.readAllLines(existing, StandardCharsets.UTF_8), theirs); + + givenLive(theirs.toArray(new String[0])); + assertEquals(sync(resultToken), Arrays.asList("DELETE d [John, Roe, pw2]", delta)); + } + + @Test + public void existingSnapshotOfTheCopiedMtimeThatCannotBeReadKeepsTheToken() throws Exception { + if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + throw new SkipException("An unreadable file needs POSIX permissions"); + } + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + // the bytes of the live file, so only the failed read keeps the token + final Path existing = givenFile(csvFile.getName() + "." + csvFile.lastModified(), HEADER, "k,Ann,Poe,pw3"); + Files.setPosixFilePermissions(existing, PosixFilePermissions.fromString("---------")); + if (Files.isReadable(existing)) { + throw new SkipException("A file without read permission can still be read by this user"); + } + connector(); + + final String output = captureStdOut(() -> assertEquals(sync(SNAPSHOT_TOKEN), + Collections.singletonList("DELETE d [John, Roe, pw2]"))); + + assertEquals(resultToken, SNAPSHOT_TOKEN); + assertTrue(output.contains("Could not compare " + existing.toFile()), output); + } + @Test public void failedSyncKeepsTheExistingSnapshotOfTheCopiedMtime() throws Exception { givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); From a2acb126ef3ca5e5b710d482feaf5ad1e6166a40 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Wed, 21 Oct 2026 12:17:54 +0300 Subject: [PATCH 10/14] [#174] Make the queued-sync race pin deterministic for a delete with no lock held The fair lock let the queued sync take the write lock first, which pins a delete taken back under the write lock, but a delete with no lock held raced with the queued sync and survived most runs. The test now swaps in a lock whose read-lock release waits until the queued sync has taken its baseline, so both orders fail on every run. --- .../openicf/csvfile/SyncOpDiffTest.java | 61 ++++++++++++++----- 1 file changed, 46 insertions(+), 15 deletions(-) diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index deac4af0..3cb442f3 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -42,6 +42,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -533,16 +534,46 @@ File copyCsvFile() throws IOException { public void syncQueuedBehindAFailingSyncGetsATokenWithASnapshot() throws Exception { givenLive(HEADER, "u1,Jane,Doe,pw1"); connector(); - // a fair lock hands the write lock to the queued sync before the failing sync can take it back - final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true); - lockMap().put(csvFile.getAbsolutePath(), lock); - final ExecutorService other = Executors.newSingleThreadExecutor(); + final Thread failing = Thread.currentThread(); final List> queued = new ArrayList<>(); + final CountDownLatch baselineTaken = new CountDownLatch(1); + // once the failing sync releases the read lock, it waits until the queued sync has taken its baseline and + // released the write lock, so a snapshot the failing sync deletes only after the release is already reused + lockMap().put(csvFile.getAbsolutePath(), new ReentrantReadWriteLock() { + private final ReadLock readLock = new ReadLock(this) { + @Override + public void unlock() { + super.unlock(); + if (Thread.currentThread() == failing && !queued.isEmpty()) { + await(baselineTaken); + } + } + }; + + private final WriteLock writeLock = new WriteLock(this) { + @Override + public void unlock() { + super.unlock(); + if (Thread.currentThread() != failing) { + baselineTaken.countDown(); + } + } + }; + + @Override + public ReadLock readLock() { + return readLock; + } + + @Override + public WriteLock writeLock() { + return writeLock; + } + }); + final ExecutorService other = Executors.newSingleThreadExecutor(); try { - expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, () -> { - queued.add(other.submit(() -> sync(null))); - awaitQueued(lock); - })); + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, + () -> queued.add(other.submit(() -> sync(null))))); await(queued.get(0)); // the old code deleted the snapshot of the failing sync after the queued sync had reused it @@ -1163,14 +1194,14 @@ private static T await(Future task) { } } - // waits until another thread queues for the lock - private static void awaitQueued(ReentrantReadWriteLock lock) { - final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); - while (!lock.hasQueuedThreads()) { - if (System.nanoTime() > deadline) { - throw new AssertionError("No thread queued for " + lock); + // waits for another thread to count the latch down + private static void await(CountDownLatch latch) { + try { + if (!latch.await(10, TimeUnit.SECONDS)) { + throw new AssertionError("Timed out waiting for " + latch); } - Thread.yield(); + } catch (InterruptedException e) { + throw new AssertionError(e); } } From ce1ca89a5faed14c6724117cf7efd8bd0b7e4cba Mon Sep 17 00:00:00 2001 From: maximthomas Date: Fri, 23 Oct 2026 11:20:33 +0300 Subject: [PATCH 11/14] [#174] Scrub CSV sync snapshots and pick the latest token in the order they were taken The scrub deleted the snapshots with the lowest mtime in their names first. Those are the oldest only while the CSV file's mtime moves forward. After a restore from a backup, cp -p or rsync -t, the sync that reports the change names its snapshot with the lowest mtime, and the next scrub of another consumer deleted it, so the next sync from that token failed. A snapshot is copied without the attributes of the CSV file and renamed into place, so its own mtime is the time of its copy. listSnapshots() now orders snapshots by that mtime, with a stable sort over the name order, so snapshots taken in one tick of a coarse clock keep the order of their names. getLatestSyncToken returns the token of the snapshot taken last, which the scrub keeps longest, instead of the newest name, which the scrub could now delete first. --- .../openicf/csvfile/CSVFileConnector.java | 36 ++++++++---- .../openicf/csvfile/SyncOpDiffTest.java | 57 +++++++++++++++++++ 2 files changed, 81 insertions(+), 12 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index 334fe479..eb730a3c 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -728,9 +728,10 @@ public SyncToken getLatestSyncToken(ObjectClass objectClass) { final ReadLock lock = fileNameToLockMap.get(csvFilePath).readLock(); lock.lock(); try { - final TreeMap snapshots = listSnapshots(); + final List> snapshots = listSnapshots(); if (!snapshots.isEmpty()) { - return new SyncToken(snapshots.lastKey()); + // the snapshot taken last, which the scrub keeps longest + return new SyncToken(snapshots.get(snapshots.size() - 1).getKey()); } // every token handed out has a snapshot, so the first sync from it reports every change made since final long timestamp = config.getCsvFile().lastModified(); @@ -1051,40 +1052,51 @@ private File snapshotFile(Object mtime) { } /** - * Lists the sync snapshots of the CSV file, {@code .<13-digit mtime>}, oldest first. The name is matched - * exactly, so the snapshots of another CSV file in the same directory are never listed. + * Lists the sync snapshots of the CSV file, {@code .<13-digit mtime>}, each by its token, the mtime in its + * name, in the order they were taken. The name is matched exactly, so the snapshots of another CSV file in the + * same directory are never listed. A snapshot is a copy made without the attributes of the CSV file and renamed + * into place, so its own mtime is the time of its copy; its name follows that order only as long as the mtime of + * the CSV file never moves back, as it does on a restore from a backup. Snapshots of one own mtime, such as those + * taken in one tick of a coarse file system clock, keep the order of their names. */ - private TreeMap listSnapshots() { + private List> listSnapshots() { final File dataDir = config.getCsvFile().getAbsoluteFile().getParentFile(); final String[] names = dataDir.list(); if (names == null) { throw new ConnectorIOException("Unable to list the files of " + dataDir); } final Pattern snapshotName = Pattern.compile(Pattern.quote(config.getCsvFile().getName()) + "\\.([0-9]{13})"); - final TreeMap snapshots = new TreeMap(); + final TreeMap byName = new TreeMap(); for (String name : names) { final Matcher matcher = snapshotName.matcher(name); if (matcher.matches()) { - snapshots.put(Long.valueOf(matcher.group(1)), new File(dataDir, name)); + byName.put(Long.valueOf(matcher.group(1)), new File(dataDir, name)); } } + final Map taken = new HashMap(); + for (File snapshot : byName.values()) { + taken.put(snapshot, snapshot.lastModified()); + } + final List> snapshots = new ArrayList>(byName.entrySet()); + // a stable sort, so the snapshots of one own mtime stay in the order of their names + Collections.sort(snapshots, (a, b) -> Long.compare(taken.get(a.getValue()), taken.get(b.getValue()))); return snapshots; } /** - * Deletes the snapshots older than the {@code syncFileRetentionCount} newest ones, except the snapshot this sync + * Deletes the snapshots taken before the {@code syncFileRetentionCount} last ones, except the snapshot this sync * reads. Runs under the write lock, so no sync in this JVM reads a snapshot while it is deleted; one in another * JVM may. A snapshot that cannot be deleted, such as one still open on Windows, is logged and kept. */ private void scrubSyncFiles(File syncOrigin) { - final TreeMap snapshots = listSnapshots(); + final List> snapshots = listSnapshots(); int surplus = snapshots.size() - config.getSyncFileRetentionCount(); - for (File snapshot : snapshots.values()) { + for (Map.Entry snapshot : snapshots) { if (surplus-- <= 0) { break; } - if (!snapshot.equals(syncOrigin)) { - deleteRewriteFile(snapshot); + if (!snapshot.getValue().equals(syncOrigin)) { + deleteRewriteFile(snapshot.getValue()); } } } diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index 3cb442f3..4bc3bc33 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -765,6 +765,43 @@ public void syncKeepsTheSnapshotItReadsWhenItIsTheOldest() throws Exception { csvFile.getName() + ".1500000000000", csvFile.getName() + "." + resultToken.getValue())); } + @Test + public void syncKeepsTheTokenOfAnotherConsumerWhenTheCsvFileMtimeMovesBack() throws Exception { + retentionCount = 2; + // the tokens of consumers B and A, each taken while the CSV file had the mtime in its name + final SyncToken tokenOfA = new SyncToken(1400000000000L); + givenSnapshotTakenAt(SNAPSHOT_TOKEN, HEADER, "k,Ann,Poe,pw3"); + givenSnapshotTakenAt(tokenOfA, HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + // then the CSV file is replaced by one with an older mtime, such as by a restore from a backup + givenLive(HEADER, "k,Ann,Poe,pw3"); + Files.setLastModifiedTime(csvFile.toPath(), FileTime.fromMillis(1200000000000L)); + + assertEquals(sync(tokenOfA), Collections.singletonList("DELETE d [John, Roe, pw2]")); + final SyncToken nextOfA = resultToken; + assertEquals(nextOfA.getValue(), 1200000000000L); + // B's scrub spares the snapshot it reads, the one taken first; ordered by name, it deleted A's + assertEquals(sync(SNAPSHOT_TOKEN), Collections.emptyList()); + assertEquals(snapshots().size(), 3); + + assertEquals(sync(nextOfA), Collections.emptyList()); + } + + @Test + public void snapshotsTakenInOneTickAreScrubbedInTheOrderOfTheirNames() throws Exception { + retentionCount = 2; + // a file system clock that ticks once a second or slower gives snapshots taken in a row one mtime + final FileTime tick = FileTime.fromMillis(1500000000000L); + for (Object mtime : Arrays.asList(1100000000000L, 1200000000000L, SNAPSHOT_TOKEN.getValue())) { + Files.setLastModifiedTime(givenFile(csvFile.getName() + "." + mtime, HEADER, "k,Ann,Poe,pw3"), tick); + } + givenLive(HEADER, "k,Ann,Poe,pw3"); + + assertEquals(sync(SNAPSHOT_TOKEN), Collections.emptyList()); + + assertEquals(snapshots(), Arrays.asList(csvFile.getName() + ".1200000000000", + csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue())); + } + @DataProvider public Object[][] retentionCountsBelowOne() { return new Object[][] { { 0 }, { -1 } }; @@ -876,6 +913,20 @@ public void latestSyncTokenWithoutASnapshotTakesOne() throws Exception { assertEquals(sync(token), Collections.singletonList("CREATE c [Jane, Doe, pw1]")); } + @Test + public void latestSyncTokenIsThatOfTheSnapshotTakenLast() throws Exception { + givenSnapshotTakenAt(new SyncToken(1400000000000L), HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + // taken now, after the CSV file was replaced by one with an older mtime + givenFile(csvFile.getName() + ".1200000000000", HEADER, "k,Ann,Poe,pw3"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + + final SyncToken token = connector().getLatestSyncToken(ObjectClass.ACCOUNT); + + // the token of the newest name reported a delete made before the call, and the scrub deleted it first + assertEquals(token.getValue(), 1200000000000L); + assertEquals(sync(token), Collections.emptyList()); + } + @Test public void snapshotThatCannotBeMovedInFailsSync() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { @@ -1118,6 +1169,12 @@ private Path givenFile(String name, String... lines) throws IOException { return Files.write(dir.resolve(name), Arrays.asList(lines), StandardCharsets.UTF_8); } + // a snapshot gets the time of its copy as its mtime, here the mtime of the CSV file that its name holds + private void givenSnapshotTakenAt(SyncToken token, String... lines) throws IOException { + Files.setLastModifiedTime(givenFile(csvFile.getName() + "." + token.getValue(), lines), + FileTime.fromMillis((Long) token.getValue())); + } + // the connector logs through StdOutLogger, which reads System.out on every call private static String captureStdOut(Runnable action) { final PrintStream original = System.out; From 3cefff46d1d8d6ac8edd3a08e46b0e2b2d0acb1d Mon Sep 17 00:00:00 2001 From: maximthomas Date: Fri, 23 Oct 2026 11:20:53 +0300 Subject: [PATCH 12/14] [#174] Release the read lock of a failed CSV sync even if deleting its snapshots throws The read lock was released after the loop that deletes the snapshots of a failed sync, outside a finally block. An Error or a SecurityException from that loop left it held, and every later write and sync of that CSV file in this JVM then waited for the write lock for good. The release now runs in a finally block of its own, still after the loop, and the copy for the diff is deleted after it as before. --- .../openicf/csvfile/CSVFileConnector.java | 24 +++++++----- .../openicf/csvfile/SyncOpDiffTest.java | 38 +++++++++++++++++++ 2 files changed, 52 insertions(+), 10 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index eb730a3c..39a54745 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -703,17 +703,21 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } completed = true; } finally { - if (!completed) { - // a kept copy could be handed out by getLatestSyncToken; deleted before the read lock is released, - // which, like the write lock, keeps every other sync in this JVM out, as one could reuse the copy - for (File copy : copies) { - deleteRewriteFile(copy); + try { + if (!completed) { + // a kept copy could be handed out by getLatestSyncToken; deleted before the read lock is + // released, which, like the write lock, keeps every other sync in this JVM out, as one could + // reuse the copy + for (File copy : copies) { + deleteRewriteFile(copy); + } + } + } finally { + rwLock.readLock().unlock(); + if (next != null) { + // not a snapshot, so no other sync reads it + deleteRewriteFile(next); } - } - rwLock.readLock().unlock(); - if (next != null) { - // not a snapshot, so no other sync reads it - deleteRewriteFile(next); } } } diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index 4bc3bc33..cc7847f0 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -23,6 +23,7 @@ import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; +import java.io.OutputStream; import java.io.PrintStream; import java.io.UncheckedIOException; import java.lang.reflect.Field; @@ -332,6 +333,43 @@ public void snapshotThatCannotBeDeletedIsLogged() throws Exception { assertTrue(output.contains("Could not delete " + copy), output); } + @Test + public void failedSyncReleasesTheReadLockWhenItsCleanupFails() throws Exception { + if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + throw new SkipException("A read-only directory needs POSIX permissions"); + } + final Set mode = Files.getPosixFilePermissions(dir); + final Set 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"); + connector(); + + final PrintStream original = System.out; + try { + // the snapshot this sync copied cannot be deleted, and the WARN about it throws, standing in for an Error + expectThrows(UncheckedIOException.class, () -> syncWithFailingHandleResult(null, () -> { + setDirMode(readOnly); + System.setOut(new PrintStream(new OutputStream() { + @Override + public void write(int b) { + throw new UncheckedIOException(new IOException("Standard output is closed")); + } + })); + })); + } finally { + System.setOut(original); + setDirMode(mode); + } + + // otherwise every later write and sync of this CSV file in this JVM waits for the write lock for good + assertEquals(lockMap().get(csvFile.getAbsolutePath()).getReadLockCount(), 0); + } + @Test public void syncThatCannotListItsSnapshotsDeletesTheSnapshotItCopied() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { From 2ada596024af689a10a8fafc37fddd3d218817b9 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Fri, 23 Oct 2026 11:21:46 +0300 Subject: [PATCH 13/14] [#174] Log a CSV snapshot that is gone before the byte comparison The length check ran before the try block, and File.length() returns 0 for a missing file, so a snapshot deleted meanwhile in another JVM was rejected without the "Could not compare" WARN. The sizes are now compared after both files are opened, where a missing snapshot throws NoSuchFileException and is logged. --- .../openicf/csvfile/CSVFileConnector.java | 8 ++++---- .../openicf/csvfile/SyncOpDiffTest.java | 20 +++++++++++++++++++ 2 files changed, 24 insertions(+), 4 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index 39a54745..834fe0dc 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -1384,14 +1384,14 @@ private boolean takeSnapshot(File snapshot) throws IOException { /** * Whether an existing snapshot holds the same bytes as a copy of the CSV file. A snapshot that cannot be read, - * such as one deleted meanwhile in another JVM, does not. + * such as one deleted meanwhile in another JVM, does not, and is logged. */ private static boolean sameBytes(File snapshot, File copy) { - if (snapshot.length() != copy.length()) { - return false; - } try (InputStream in = Files.newInputStream(snapshot.toPath()); InputStream other = Files.newInputStream(copy.toPath())) { + if (Files.size(snapshot.toPath()) != Files.size(copy.toPath())) { + return false; + } final byte[] buffer = new byte[8192]; final byte[] otherBuffer = new byte[buffer.length]; int length; diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index cc7847f0..f594a489 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -694,6 +694,26 @@ public void existingSnapshotOfTheCopiedMtimeThatCannotBeReadKeepsTheToken() thro assertTrue(output.contains("Could not compare " + existing.toFile()), output); } + @Test + public void existingSnapshotOfTheCopiedMtimeDeletedMeanwhileIsLogged() throws Exception { + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + // a link to no file holds the name, so the move finds the snapshot, which then reads as deleted meanwhile + final Path existing = dir.resolve(csvFile.getName() + "." + csvFile.lastModified()); + try { + Files.createSymbolicLink(existing, dir.resolve("gone")); + } catch (UnsupportedOperationException | IOException e) { + throw new SkipException("A symbolic link cannot be created here", e); + } + connector(); + + final String output = captureStdOut(() -> assertEquals(sync(SNAPSHOT_TOKEN), + Collections.singletonList("DELETE d [John, Roe, pw2]"))); + + assertEquals(resultToken, SNAPSHOT_TOKEN); + assertTrue(output.contains("Could not compare " + existing.toFile()), output); + } + @Test public void failedSyncKeepsTheExistingSnapshotOfTheCopiedMtime() throws Exception { givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); From 6bf5b0c69fe39ebf336c4d7f2dc0f906fc2e71e6 Mon Sep 17 00:00:00 2001 From: maximthomas Date: Fri, 23 Oct 2026 11:58:00 +0300 Subject: [PATCH 14/14] [#174] Give each CSV snapshot the time of its copy as its mtime on Windows too The scrub and getLatestSyncToken order snapshots by their own mtime, taken to be the time of their copy. Files.copy gives a copy that mtime on Linux and macOS only: on Windows it calls CopyFileEx, which keeps the last-write time of the CSV file, so the order fell back to the mtime in the names. Both Windows CI jobs failed syncKeepsTheTokenOfAnotherConsumerWhenTheCsvFileMtimeMovesBack: after the CSV file's mtime moved back, the scrub of another consumer deleted the snapshot just handed out as a token. Each copy for a snapshot now gets the current time as its mtime before it is renamed into place. A copy whose mtime cannot be set is logged as a WARN and used as it is, so a copy deleted meanwhile still fails the read that follows, as before. --- .../openicf/csvfile/CSVFileConnector.java | 30 ++++++++-- .../openicf/csvfile/SyncOpDiffTest.java | 58 ++++++++++++++++++- 2 files changed, 80 insertions(+), 8 deletions(-) diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index 834fe0dc..32cb77a4 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -37,6 +37,7 @@ import java.nio.file.Path; import java.nio.file.StandardCopyOption; import java.nio.file.StandardOpenOption; +import java.nio.file.attribute.FileTime; import java.nio.file.attribute.PosixFilePermission; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; @@ -518,7 +519,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha // change would be kept over the fresh copy of that mtime, and every sync would report the change again final long nextTimestamp = config.getCsvFile().lastModified(); try { - next = copyCsvFile(); + next = copyCsvFileForSnapshot(); } catch (NoSuchFileException e) { log.error(e, "File {0} does not exist!", config.getCsvFile().toString()); throw new ConnectorIOException("File " + config.getCsvFile().toString() + " does not exist", e); @@ -1058,10 +1059,10 @@ private File snapshotFile(Object mtime) { /** * Lists the sync snapshots of the CSV file, {@code .<13-digit mtime>}, each by its token, the mtime in its * name, in the order they were taken. The name is matched exactly, so the snapshots of another CSV file in the - * same directory are never listed. A snapshot is a copy made without the attributes of the CSV file and renamed - * into place, so its own mtime is the time of its copy; its name follows that order only as long as the mtime of - * the CSV file never moves back, as it does on a restore from a backup. Snapshots of one own mtime, such as those - * taken in one tick of a coarse file system clock, keep the order of their names. + * same directory are never listed. A snapshot is a copy given the time of its copy as its mtime and renamed into + * place, see {@link #copyCsvFileForSnapshot()}; its name follows that order only as long as the mtime of the CSV + * file never moves back, as it does on a restore from a backup. Snapshots of one own mtime, such as those taken in + * one tick of a coarse file system clock, keep the order of their names. */ private List> listSnapshots() { final File dataDir = config.getCsvFile().getAbsoluteFile().getParentFile(); @@ -1356,6 +1357,23 @@ File copyCsvFile() throws IOException { } } + /** + * Copies the CSV file for a sync snapshot, with the time of the copy as its mtime, which orders the snapshots in + * {@link #listSnapshots()}. {@code Files.copy} gives a copy that mtime on Linux and macOS only: on Windows its + * {@code CopyFileEx} keeps the mtime of the CSV file. A copy whose mtime cannot be set is logged and used as it is; + * one deleted meanwhile fails the read that follows. + */ + private File copyCsvFileForSnapshot() throws IOException { + final File copy = copyCsvFile(); + try { + Files.setLastModifiedTime(copy.toPath(), FileTime.fromMillis(System.currentTimeMillis())); + } catch (IOException e) { + log.warn(e, "Could not set the mtime of copy {0} of {1} to the time of the copy", copy, + config.getCsvFile()); + } + return copy; + } + /** * Takes a snapshot of the CSV file: a complete copy, renamed into place, so that no snapshot is ever seen * half-written. Makes nothing if a snapshot of that name exists, as it holds the same mtime of the CSV file. @@ -1366,7 +1384,7 @@ private boolean takeSnapshot(File snapshot) throws IOException { if (snapshot.exists()) { return false; } - final File copy = copyCsvFile(); + final File copy = copyCsvFileForSnapshot(); boolean moved = false; try { Files.move(copy.toPath(), snapshot.toPath()); diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index f594a489..61b2fb58 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -823,8 +823,14 @@ public void syncKeepsTheSnapshotItReadsWhenItIsTheOldest() throws Exception { csvFile.getName() + ".1500000000000", csvFile.getName() + "." + resultToken.getValue())); } - @Test - public void syncKeepsTheTokenOfAnotherConsumerWhenTheCsvFileMtimeMovesBack() throws Exception { + @DataProvider + public Object[][] copiesKeepingTheCsvFileMtime() { + return new Object[][] { { false }, { true } }; + } + + @Test(dataProvider = "copiesKeepingTheCsvFileMtime") + public void syncKeepsTheTokenOfAnotherConsumerWhenTheCsvFileMtimeMovesBack(boolean copyKeepsTheCsvFileMtime) + throws Exception { retentionCount = 2; // the tokens of consumers B and A, each taken while the CSV file had the mtime in its name final SyncToken tokenOfA = new SyncToken(1400000000000L); @@ -833,7 +839,11 @@ public void syncKeepsTheTokenOfAnotherConsumerWhenTheCsvFileMtimeMovesBack() thr // then the CSV file is replaced by one with an older mtime, such as by a restore from a backup givenLive(HEADER, "k,Ann,Poe,pw3"); Files.setLastModifiedTime(csvFile.toPath(), FileTime.fromMillis(1200000000000L)); + if (copyKeepsTheCsvFileMtime) { + connectorWhoseCopiesKeepTheCsvFileMtime(); + } + // A's sync takes the snapshot of its next token while it reads the changes assertEquals(sync(tokenOfA), Collections.singletonList("DELETE d [John, Roe, pw2]")); final SyncToken nextOfA = resultToken; assertEquals(nextOfA.getValue(), 1200000000000L); @@ -985,6 +995,25 @@ public void latestSyncTokenIsThatOfTheSnapshotTakenLast() throws Exception { assertEquals(sync(token), Collections.emptyList()); } + @Test(dataProvider = "copiesKeepingTheCsvFileMtime") + public void latestSyncTokenIsThatOfABaselineTakenAfterTheCsvFileMtimeMovesBack(boolean copyKeepsTheCsvFileMtime) + throws Exception { + givenSnapshotTakenAt(new SyncToken(1400000000000L), HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + Files.setLastModifiedTime(csvFile.toPath(), FileTime.fromMillis(1200000000000L)); + if (copyKeepsTheCsvFileMtime) { + connectorWhoseCopiesKeepTheCsvFileMtime(); + } + // a sync without a token takes the baseline of the restored CSV file + assertEquals(sync(null), Collections.emptyList()); + + final SyncToken token = connector().getLatestSyncToken(ObjectClass.ACCOUNT); + + // a baseline with the mtime of the CSV file sorts before the older snapshot, whose token reports a delete + assertEquals(token.getValue(), 1200000000000L); + assertEquals(sync(token), Collections.emptyList()); + } + @Test public void snapshotThatCannotBeMovedInFailsSync() throws Exception { if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { @@ -1198,6 +1227,18 @@ public void copyDeletedBeforeTheIndexPassFailsSyncNamingTheCsvFile() throws Exce assertTrue(output.contains("Copy " + copies.get(0) + " of " + csvFile + " does not exist!"), output); } + @Test + public void copyDeletedBeforeItsMtimeIsSetLogsAWarning() throws Exception { + givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); + givenLive(HEADER, "k,Ann,Poe,pw3"); + final List copies = connectorRecordingCopies(true); + + final String output = + captureStdOut(() -> expectThrows(ConnectorIOException.class, () -> sync(SNAPSHOT_TOKEN))); + + assertTrue(output.contains("Could not set the mtime of copy " + copies.get(0) + " of " + csvFile), output); + } + @Test public void copyDeletedBeforeTheCreateUpdatePassFailsSyncNamingTheCsvFile() throws Exception { givenSnapshot(HEADER, "k,Ann,Poe,pw3", "d,John,Roe,pw2"); @@ -1359,6 +1400,19 @@ File copyCsvFile() throws IOException { return copies; } + // copies the CSV file as Files.copy does on Windows, where CopyFileEx keeps the mtime of the source file; on Linux + // and macOS a copy gets the time it is made + private void connectorWhoseCopiesKeepTheCsvFileMtime() { + connector(new CSVFileConnector() { + @Override + File copyCsvFile() throws IOException { + final File copy = super.copyCsvFile(); + Files.setLastModifiedTime(copy.toPath(), Files.getLastModifiedTime(csvFile.toPath())); + return copy; + } + }); + } + private CSVFileConnector connector(CSVFileConnector instance) { CSVFileConfiguration config = new CSVFileConfiguration(); config.setCsvFile(csvFile);