Reject storing and committing after a failed store until clear()

persistence/binary/src/main/java/org/eclipse/serializer/persistence/binary/types/BinaryStorer.java

@@ -197,6 +197,13 @@ protected static int defaultSlotSize()
 		 */
 		protected boolean isProcessingItems;
 
+		/*
+		 * First failure of item processing. A failed store leaves partially serialized data (unprocessed items,
+		 * a half-written entity, commit listeners of the failed store), so the storer must not store or commit
+		 * until it is cleared. Guarded by objectRegistryMonitor.
+		 */
+		private Throwable storeFailure;
+
 		/*
 		 * item hashing structures get initialized lazily for the following reasons:
 		 * - the storer instance can commit (be cleared) and be reinitialized multiple times.
@@ -428,6 +435,8 @@ protected void internalInitialize(final int hashLength)
 				this.hashSlots = new Item[hashLength];
 				this.hashRange = hashLength - 1;
 
+				this.storeFailure = null;
+
 				// initializing/clearing item chain
 				(this.tail = this.head).next = null;
 
@@ -603,6 +612,11 @@ protected long internalStore(final Object root)
 			 * - handle all registered graph objects recursively (but transformed to an iteration).
 			 * Note that this is NOT the same as apply, which does NOT store if the instance is already registry-known.
 			 */
+			if(!this.isProcessingItems)
+			{
+				this.validateNoFailedStore();
+			}
+
 			long rootOid;
 			if(Swizzling.isFoundId(rootOid = this.lookupOid(root)))
 			{
@@ -615,15 +629,7 @@ protected long internalStore(final Object root)
 			// repeatedly calling #store to add an instance to the item chain is fine, but processing may only happen once.
 			if(!this.isProcessingItems)
 			{
-				try
-				{
-					this.isProcessingItems = true;
-					this.processItems();
-				}
-				finally
-				{
-					this.isProcessingItems = false;
-				}
+				this.processItemsOutermost();
 			}
 
 			return rootOid;
@@ -666,6 +672,65 @@ protected final void storeItem(final Item item)
 			item.typeHandler.store(chunk, item.instance, item.oid, this);
 		}
 
+		/**
+		 * Processes the item chain as the outermost store call and records the first failure.
+		 */
+		protected final void processItemsOutermost()
+		{
+			try
+			{
+				this.isProcessingItems = true;
+				this.processItems();
+			}
+			catch(final Throwable t)
+			{
+				synchronized(this.objectRegistryMonitor)
+				{
+					if(this.storeFailure == null)
+					{
+						this.storeFailure = t;
+					}
+				}
+				throw t;
+			}
+			finally
+			{
+				this.isProcessingItems = false;
+			}
+		}
+
+		/**
+		 * Rejects storing and committing after a failed store: committing would persist the partially
+		 * serialized graph. Only {@link #clear()} resets the storer.
+		 * <p>
+		 * Must be called under {@code objectRegistryMonitor} and not during item processing.
+		 */
+		protected final void synchValidateNoFailedStore()
+		{
+			if(this.storeFailure != null)
+			{
+				throw new PersistenceException(
+					"A store of this storer failed and left partially serialized data."
+					+ " Committing it would persist an inconsistent object graph."
+					+ " The storer must be cleared before it can be used again.",
+					this.storeFailure
+				);
+			}
+		}
+
+		protected final Throwable synchStoreFailure()
+		{
+			return this.storeFailure;
+		}
+
+		protected final void validateNoFailedStore()
+		{
+			synchronized(this.objectRegistryMonitor)
+			{
+				this.synchValidateNoFailedStore();
+			}
+		}
+
 		@Override
 		public final long store(final Object root)
 		{
@@ -797,6 +862,9 @@ public Object commit()
 				LazyArg(this::size)   // use lazy here, #size() locks
 			);
 
+			// before anything is written, merged or notified: the storer keeps its state until cleared
+			this.validateNoFailedStore();
+
 			// isEmpty locks internally
 			if(!this.isEmpty())
 			{
@@ -1455,6 +1523,11 @@ protected final long internalStore(final Object root, final long objectId)
 			 * - handle all registered graph objects recursively (but transformed to an iteration).
 			 * Note that this is NOT the same as apply, which does NOT store if the instance is already registry-known.
 			 */
+			if(!this.isProcessingItems)
+			{
+				this.validateNoFailedStore();
+			}
+
 			final long rootOid;
 			if(Swizzling.isFoundId(rootOid = this.lookupOid(root)))
 			{
@@ -1469,15 +1542,7 @@ protected final long internalStore(final Object root, final long objectId)
 			// repeatedly calling #store to add an instance to the item chain is fine, but processing may only happen once.
 			if(!this.isProcessingItems)
 			{
-				try
-				{
-					this.isProcessingItems = true;
-					this.processItems();
-				}
-				finally
-				{
-					this.isProcessingItems = false;
-				}
+				this.processItemsOutermost();
 			}
 
 			return objectId;
@@ -1490,6 +1555,11 @@ public void forceRootStore(final PersistenceRoots pendingStoreRoot) {
 
 			logger.debug("Storing updated root");
 
+			if(!this.isProcessingItems)
+			{
+				this.validateNoFailedStore();
+			}
+
 			long rootOid;
 			if(Swizzling.isFoundId(rootOid = this.lookupOid(pendingStoreRoot)))
 			{
@@ -1502,15 +1572,7 @@ public void forceRootStore(final PersistenceRoots pendingStoreRoot) {
 
 			if(!this.isProcessingItems)
 			{
-				try
-				{
-					this.isProcessingItems = true;
-					this.processItems();
-				}
-				finally
-				{
-					this.isProcessingItems = false;
-				}
+				this.processItemsOutermost();
 			}
 		}
 
@@ -1681,6 +1743,7 @@ public final class Batching extends Default implements BatchStorer
 		private final BatchStorer.Controller   controller        ;
 		private final BackgroundFlusher        flusher           ;
 		private long                           pendingSinceNanos ;
+		private boolean                        failureReported   ;
 		private volatile boolean               closed            ;
 
 		/*
@@ -1761,19 +1824,15 @@ public long store(final Object instance, final long objectId)
 				{
 					throw new IllegalStateException("BatchStorer is already closed.");
 				}
+				if(!this.isProcessingItems)
+				{
+					this.validateNoFailedStore();
+				}
 				this.registerGuaranteed(objectId, instance, null);
 
 				if(!this.isProcessingItems)
 				{
-					try
-					{
-						this.isProcessingItems = true;
-						super.processItems();
-					}
-					finally
-					{
-						this.isProcessingItems = false;
-					}
+					this.processItemsOutermost();
 				}
 			}
 
@@ -1850,6 +1909,10 @@ protected long internalStore(final Object root)
 				{
 					throw new IllegalStateException("BatchStorer is already closed.");
 				}
+				if(!this.isProcessingItems)
+				{
+					this.validateNoFailedStore();
+				}
 
 				final long localOid = this.lookupOid(root);
 
@@ -1868,15 +1931,7 @@ protected long internalStore(final Object root)
 
 				if(!this.isProcessingItems)
 				{
-					try
-					{
-						this.isProcessingItems = true;
-						super.processItems();
-					}
-					finally
-					{
-						this.isProcessingItems = false;
-					}
+					this.processItemsOutermost();
 				}
 			}
 
@@ -1908,10 +1963,33 @@ protected void processItems()
 		@Override
 		public void clear()
 		{
-			synchronized(this.objectRegistryMonitor)
+			// commitLock: an in-flight store of another thread must not continue into the cleared chain and chunks
+			synchronized(this.commitLock)
+			{
+				synchronized(this.objectRegistryMonitor)
+				{
+					super.clear();
+					this.pendingSinceNanos = 0L;
+					this.failureReported   = false;
+				}
+			}
+		}
+
+		@Override
+		public PersistenceStorer reinitialize()
+		{
+			synchronized(this.commitLock)
 			{
-				super.clear();
-				this.pendingSinceNanos = 0L;
+				return super.reinitialize();
+			}
+		}
+
+		@Override
+		public PersistenceStorer reinitialize(final long initialCapacity)
+		{
+			synchronized(this.commitLock)
+			{
+				return super.reinitialize(initialCapacity);
 			}
 		}
 
@@ -1995,6 +2073,18 @@ private void optFlush()
 					return;
 				}
 
+				// a failed store blocks committing until clear(): report it once instead of failing on every tick
+				final Throwable storeFailure = this.synchStoreFailure();
+				if(storeFailure != null)
+				{
+					if(!this.failureReported)
+					{
+						this.failureReported = true;
+						logger.error("Batch not flushed: a store failed, the pending data is kept until clear()", storeFailure);
+					}
+					return;
+				}
+
 				final long nowNanos = System.nanoTime();
 				if(this.pendingSinceNanos == 0L)
 				{

persistence/persistence/src/main/java/org/eclipse/serializer/persistence/types/BatchStorer.java

@@ -36,6 +36,14 @@
  * {@code BatchStorer} implements {@link AutoCloseable}. On close, it flushes any remaining
  * pending data and releases resources.
  * <p>
+ * If a store call fails, the batch holds incomplete data and is not flushed: further store calls,
+ * {@link #flush()}, {@link #commit()} and {@link #close()} throw a
+ * {@link org.eclipse.serializer.persistence.exceptions.PersistenceException}, and the background
+ * flush skips the batch. This affects the whole batch, including data stored by other threads.
+ * {@link #clear()} discards the pending data and makes the {@code BatchStorer} usable again, unless
+ * it was already closed: {@link #close()} releases its resources even when the final flush is
+ * rejected, so a failed batch must be cleared before closing it to keep using it.
+ * <p>
  * Usage:
  * <pre>
  * try (BatchStorer storer = storageManager.batchStorerBuilder()

persistence/persistence/src/main/java/org/eclipse/serializer/persistence/types/Storer.java

@@ -52,6 +52,11 @@ public interface Storer extends PersistenceStoring
 	 * <p>
 	 * The method may be called at most once on a fresh storer state. To reuse the storer, call
 	 * {@link #clear()} or {@link #reinitialize()} afterwards.
+	 * <p>
+	 * If a {@code store(...)} call on this storer failed, the accumulated data is incomplete. In that
+	 * case {@code commit()} persists nothing and throws a
+	 * {@link org.eclipse.serializer.persistence.exceptions.PersistenceException} with the store
+	 * failure as its cause, until {@link #clear()} discards the accumulated data.
 	 *
 	 * @return implementation-specific status information about the commit, or {@code null} if the
 	 *         implementation does not provide any.
@@ -69,6 +74,8 @@ public interface Storer extends PersistenceStoring
 	 * This rolls back uncommitted in-memory state only; data already committed by a previous
 	 * {@link #commit()} call is unaffected. After {@code clear()} the storer is ready to start a new
 	 * collection cycle.
+	 * <p>
+	 * After a failed {@code store(...)} call, {@code clear()} is the only way to use the storer again.
 	 *
 	 * @see #commit()
 	 * @see #reinitialize()
@@ -264,6 +271,12 @@ public default boolean isEmpty()
 	 * {@link PersistenceObjectRegistry} and nothing is written to the underlying storage layer.
 	 * Calling {@code store(...)} on the same instance multiple times before {@code commit()} has no
 	 * additional effect.
+	 * <p>
+	 * If this call fails, for example because the graph contains an instance that cannot be persisted,
+	 * the data accumulated by this storer is incomplete. Further {@code store(...)} and
+	 * {@link #commit()} calls then throw a
+	 * {@link org.eclipse.serializer.persistence.exceptions.PersistenceException} until
+	 * {@link #clear()} is called.
 	 *
 	 * @see #commit()
 	 * @see #skip(Object)

serializer/src/main/java/org/eclipse/serializer/Serializer.java

@@ -207,8 +207,17 @@ public static class Default<M> implements Serializer<M>
 		@Override
 		public synchronized M serialize(final Object object)
 		{
-			this.storer.store(object);
-			this.storer.commit();
+			try
+			{
+				this.storer.store(object);
+				this.storer.commit();
+			}
+			catch(final Throwable t)
+			{
+				// the internal storer would carry the partial graph into the next serialize() call
+				this.storer.clear();
+				throw t;
+			}
 			return this.toMedium.apply(this.output);
 		}