Merge pull request #918 from FgForrest/dev

fix: error during closing session after goLive operation via gRPC

Author: novoj

evita_api/src/main/java/io/evitadb/api/GoLiveProgressRecord.java

@@ -101,8 +101,10 @@ public void updatePercentCompleted(int percentage) {
 		if (percentage < 0 || percentage > 100) {
 			throw new IllegalArgumentException("Percentage must be between 0 and 100, but was: " + percentage);
 		}
-		this.percentCompleted.set(percentage);
-		notifyClientObserver(percentage);
+		final int previousValue = this.percentCompleted.getAndSet(percentage);
+		if (previousValue != percentage) {
+			notifyClientObserver(percentage);
+		}
 	}
 
 	/**

evita_engine/src/main/java/io/evitadb/core/Evita.java

@@ -48,6 +48,7 @@
 import io.evitadb.api.requestResponse.schema.mutation.catalog.RemoveCatalogSchemaMutation;
 import io.evitadb.api.task.ServerTask;
 import io.evitadb.core.SessionRegistry.SuspendOperation;
+import io.evitadb.core.SessionRegistry.SuspensionInformation;
 import io.evitadb.core.async.ClientRunnableTask;
 import io.evitadb.core.async.EmptySettings;
 import io.evitadb.core.async.ObservableExecutorServiceWithHardDeadline;
@@ -326,6 +327,28 @@ public void emitStartObservabilityEvents() {
 		);
 	}
 
+	/**
+	 * Checks if sessions were forcefully closed for the specified catalog and session ID.
+	 *
+	 * @param catalogName the name of the catalog for which to check if sessions were forcefully closed; must not be null
+	 * @param sessionId the unique identifier of the session to check; must not be null
+	 * @return true if sessions were forcefully closed for the specified catalog and session ID, false otherwise
+	 */
+	public boolean wasSessionForcefullyClosedForCatalog(@Nonnull String catalogName, @Nonnull UUID sessionId) {
+		return ofNullable(this.catalogSessionRegistries.get(catalogName))
+			.map(it -> it.wereSessionsForcefullyClosedForCatalog(sessionId))
+			.orElse(false);
+	}
+
+	/**
+	 * Clears all session registries and their temporary information.
+	 */
+	public void clearSessionRegistries() {
+		for (SessionRegistry value : this.catalogSessionRegistries.values()) {
+			value.clearTemporaryInformation();
+		}
+	}
+
 	/**
 	 * Emits statistics of the ThreadPool associated with the scheduler.
 	 */
@@ -741,6 +764,29 @@ ServerTask<EmptySettings, Void> createLoadCatalogTask(@Nonnull String catalogNam
 		);
 	}
 
+	/**
+	 * Closes all active sessions associated with the specified catalog and suspends further operations.
+	 *
+	 * @param catalogName      the name of the catalog whose sessions are to be closed and suspended
+	 * @param suspendOperation the operation to be executed during the suspension of the catalog
+	 */
+	@Nonnull
+	public Optional<SuspensionInformation> closeAllSessionsAndSuspend(@Nonnull String catalogName, @Nonnull SuspendOperation suspendOperation) {
+		return ofNullable(this.catalogSessionRegistries.get(catalogName))
+			.flatMap(it -> it.closeAllActiveSessionsAndSuspend(suspendOperation));
+	}
+
+	/**
+	 * Discards the suspension state of the session registry associated with the given catalog name, if present.
+	 * The method resumes operations for the session registry if it exists for the provided catalog name.
+	 *
+	 * @param catalogName The name of the catalog whose suspension state should be discarded. Must not be null.
+	 */
+	public void discardSuspension(@Nonnull String catalogName) {
+		ofNullable(this.catalogSessionRegistries.get(catalogName))
+			.ifPresent(SessionRegistry::resumeOperations);
+	}
+
 	/*
 		PRIVATE METHODS
 	*/
@@ -962,6 +1008,9 @@ private void replaceCatalogReference(@Nonnull Catalog catalog) {
 			}
 		);
 
+		// discard suspension of the session registry for the catalog, if present
+		discardSuspension(catalogName);
+
 		// notify structural changes callbacks
 		ofNullable(originalCatalog.get())
 			.ifPresent(it -> notifyStructuralChangeObservers(catalog, it));

evita_engine/src/main/java/io/evitadb/core/EvitaSession.java

@@ -73,6 +73,7 @@
 import io.evitadb.api.task.ServerTask;
 import io.evitadb.api.task.Task;
 import io.evitadb.api.task.TaskStatus;
+import io.evitadb.core.SessionRegistry.SuspendOperation;
 import io.evitadb.core.async.Interruptible;
 import io.evitadb.core.async.Scheduler;
 import io.evitadb.core.cdc.predicate.MutationPredicateFactory;
@@ -410,6 +411,8 @@ public GoLiveProgress goLiveAndCloseWithProgress(@Nullable IntConsumer progressO
 		isTrue(!theCatalog.supportsTransaction(), "Catalog went live already and is currently in transactional mode!");
 		executeTerminationSteps(null, theCatalog);
 		this.closedFuture = CompletableFuture.completedFuture(new CommitVersions(this.catalog.getVersion() + 1, this.catalog.getSchema().version()));
+		this.evita.closeAllSessionsAndSuspend(this.catalog.getName(), SuspendOperation.REJECT)
+			.ifPresent(it -> it.addForcefullyClosedSession(this.id));
 		return theCatalog.goLive(progressObserver);
 	}