Fix ReplaceChangeLogLockStatement

[raffaeleflorio]

The current update method of ReplaceChangeLogLockStatement is not safe. It allows multiple owner (identified by the lockedBy field) to acquire the lock simultaneously. And it also allows the release by a different owner. There isn't any fence in the current implementation:

liquibase-mongodb/src/main/java/liquibase/ext/mongodb/lockservice/ReplaceChangeLogLockStatement.java

Lines 77 to 84 in 5582bcd

	final Optional<Document> changeLogLock = Optional.ofNullable(
	database.getMongoDatabase()
	.getCollection(collectionName)
	.findOneAndReplace(Filters.eq(MongoChangeLogLock.Fields.id,
	entry.getId()), inputDocument,
	new FindOneAndReplaceOptions().upsert(upsert).returnDocument(ReturnDocument.AFTER))
	);
	return changeLogLock.map(e -> 1).orElse(0);

Indeed, by running the following code with the latest liquibase-mongodb version there will be also multiple exceptions:

public final class Main {
    public static void main(final String[] args) throws Exception {
        for (var i = 0; i < 16; i++) {
            Executors.newSingleThreadExecutor().execute(() ->
                    {
                        try (
                                var db = DatabaseFactory.getInstance().openDatabase(
                                        "mongodb://localhost:27017/test?directConnection=true",
                                        null,
                                        null,
                                        null,
                                        null
                                );
                                var liquibase = new Liquibase(
                                        "db/changelog.xml",
                                        new ClassLoaderResourceAccessor(),
                                        db
                                )
                        ) {
                            liquibase.update("");
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                    }
            );
        }
        Thread.sleep(10000);
    }
}

The proposed implementation blocks multiple acquisition by taking advantage of the Mongo DuplicateKey error. Specifically, an owner will try to acquire only an unlocked lock. And we have multiple scenarios here:

1. It finds the unlocked lock and it acquires the lock by replacing the Mongo document
2. It doesn't find the lock because this is the first acquisition request ever. So, it acquires the lock by inserting the Mongo document
3. It doesn't find the lock because it's already locked. In this case, it will try in any case to insert the document in Mongo but this time it will fail to acquire because the DuplicateKey error

The same approach is used to allow the lock release only by his owner.

[raffaeleflorio]

Hi @filipelautert @KushnirykOleh, is there anything else should I improve about this PR?

[filipelautert]

Hi @filipelautert @KushnirykOleh, is there anything else should I improve about this PR?

Hi @raffaeleflorio , thanks for the PR! Would you be able to convert you example from the description to an unit/integration test?

[raffaeleflorio]

Hi @filipelautert @KushnirykOleh, is there anything else should I improve about this PR?

Hi @raffaeleflorio , thanks for the PR! Would you be able to convert you example from the description to an unit/integration test?

Hi @filipelautert I've just pushed ReplaceChangeLogLockStatementIT. With the previous implementation multipleLock and lockThenUnlockByDifferentHost fail.

I also wrote a test in the MongoLockServiceIT that resemble more the example I wrote, but because concurrency it's brittler than ReplaceChangeLogLockStatementIT. That's why I preferred to push just the ReplaceChangeLogLockStatementIT . As a reference here is the test I didn't push:

    @SneakyThrows
    @RepeatedTest(8)
    void parallelAcquireLock() {
        AtomicInteger acquiredLocks = new AtomicInteger(0);
        LinkedBlockingQueue<Future<?>> acquisitions = new LinkedBlockingQueue<>();
        for (int i = 0; i < 16; i++) {
            acquisitions.add(
                    Executors.newSingleThreadExecutor().submit(() -> {
                        try {
                            MongoLockService ls = (MongoLockService) LockServiceFactory.getInstance().getLockService(database);
                            ls.reset();
                            if (ls.acquireLock()) {
                                acquiredLocks.incrementAndGet();
                            }
                        } catch (LockException ignored) {
                        }
                    })
            );
        }
        while (acquisitions.peek() != null) {
            acquisitions.poll().get();
        }
        assertThat(acquiredLocks).hasValue(1);
    }

[filipelautert]

Just one comment related to.... comments. Besides it looks good! @KushnirykOleh could you review it too?

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud