patch:LoggedDataset: concurrent write transactions corrupt the change log ("Already writing", NPE this.tok is null, several transactions in one patch file)

Start with DatasetGraphChanges.begin(TxnType) and RDFChangesManagedOutput in the Jena source; trace how transaction start and output sections interact while a writer waits for the dataset lock. Reproduce with two concurrent write transactions and inspect the patch files and replayed dataset. Done means concurrent updates produce separate, replayable transaction sections without the reported exception or data mismatch.

Version

Jena / Fuseki 6.2.0 (official apache-jena-fuseki-6.2.0 distribution, Docker, Java 21). The code involved is unchanged on main (checked at d774384789).

What happened?

Summary

A dataset wrapped with patch:LoggedDataset (jena-rdfpatch file log, patch:logPolicy "index") served by Fuseki cannot handle two write transactions that begin at the same time. The second transaction does not get its own patch file; it writes into the first transaction's output. When the first commits, the shared output is closed and the second fails with a NullPointerException (HTTP 400 from Fuseki). Under real traffic this also leaves a patch file containing more than one transaction, and replaying the log on a replica produced a dataset that differs from the primary.

Reproduction

1. Create a dataset via POST /$/datasets with:

   :service fuseki:name "j5" ; fuseki:endpoint [ fuseki:operation fuseki:update ; fuseki:name "update" ] ; fuseki:dataset :logged .
   :logged rdf:type patch:LoggedDataset ; patch:log "/fuseki/patchlog/j5.rdfp" ; patch:logPolicy "index" ; ja:dataset :tdb .
   :tdb rdf:type tdb2:DatasetTDB2 ; tdb2:location "/fuseki/base/databases/j5" .
   

2. Send 20 INSERT DATA updates to /j5/update concurrently (one HTTP request each).

Observed (20 concurrent updates)

• 5 succeed (204), 15 fail with HTTP 400: Cannot invoke "org.apache.jena.rdfpatch.text.TokenWriter.startTuple()" because "this.tok" is null
• Fuseki log: WARN RDFChangesManagedOutput :: Already writing ×15
• Only 5 patch files are written; one of them contains more than one TX .
• Sent sequentially, the same 20 updates all succeed and produce 20 patch files.

In a longer run with an application (several writers on the same dataset, ~1 write/s), a replica built by applying the patch files in index order ended up with one extra triple compared to the primary: after an update that replaced a value (DELETE {...} INSERT {...} WHERE {...}), the replica had both the old and the new value.

The same happens in-process, without Fuseki: DatasetGraphChanges over DatasetGraphFactory.createTxnMem() with an RDFChangesManagedOutput on an index-policy OutputManagedFile, and two threads each running a write transaction, the second starting while the first is active. A JUnit test for this is part of the proposed fix.

Cause (6.2.0 source)

• DatasetGraphChanges.begin(TxnType) calls changesMonitor.txnBegin() before super.begin(txnType). The underlying TDB2 dataset serializes writers inside super.begin, so writers that are still waiting for the write lock have already started logging.
• RDFChangesManagedOutput keeps a single currentStream / tok with no synchronization. startOutput() for the second transaction sees currentStream != null, logs "Already writing" and returns without opening a new section. finishOutput() of the first transaction then sets tok = null, and the second transaction's first change throws the NPE.

Suggested fix

Keep the order in DatasetGraphChanges as it is: RDF Delta relies on txnBegin() being called first (its RDFChangesDS.txnBegin() takes the cluster lock, then the sync handler brings the dataset up to date under that lock, then the local transaction begins).

Instead, make the file log cope with txnBegin() being called by a writer that is still waiting for the write lock:

• RDFChangesManagedOutput.txnBegin() only records, for the calling thread, that a transaction is starting; it does not touch shared state.
• The section (file) is opened when the transaction first logs something (a change, or commit/abort). Only the transaction holding the dataset write lock can do that, so sections do not overlap. A committed transaction without changes still gets a TX/TC section.
• If a section is already open for another thread (a dataset that does not serialize writers), throw instead of logging "Already writing" and continuing in that section.

Alternatives considered: moving txnBegin() after super.begin() in DatasetGraphChanges breaks the RDF Delta cluster-lock protocol above; making txnBegin() wait for the file log's semaphore can deadlock with a READ_PROMOTE transaction that promotes while holding the write lock.

Workaround

Serialize writes per dataset in front of Fuseki (one in-flight update/GSP write per dataset). This only covers traffic through the proxy, not e.g. uploads in the Fuseki UI.

Relevant output and stacktrace

HTTP 400: Cannot invoke "org.apache.jena.rdfpatch.text.TokenWriter.startTuple()" because "this.tok" is null
WARN RDFChangesManagedOutput :: Already writing

Are you interested in making a pull request?

Yes