feat(client): add full session lifecycle management

Author: camory

armonik-client/src/main/java/fr/aneo/armonik/client/ArmoniKClient.java

@@ -18,13 +18,15 @@
 import fr.aneo.armonik.api.grpc.v1.sessions.SessionsGrpc;
 import fr.aneo.armonik.client.definition.SessionDefinition;
 import fr.aneo.armonik.client.exception.ArmoniKException;
+import fr.aneo.armonik.client.internal.concurrent.Futures;
 import io.grpc.ManagedChannel;
 
 import java.time.Duration;
 import java.util.HashSet;
+import java.util.concurrent.CompletionStage;
 
-import static fr.aneo.armonik.client.internal.grpc.mappers.SessionMapper.toCreateSessionRequest;
-import static fr.aneo.armonik.client.internal.grpc.mappers.SessionMapper.toGetSessionRequest;
+import static fr.aneo.armonik.client.internal.grpc.mappers.SessionMapper.*;
+import static fr.aneo.armonik.client.internal.grpc.mappers.SessionMapper.toSessionState;
 import static fr.aneo.armonik.client.internal.grpc.mappers.TaskMapper.toTaskConfiguration;
 import static java.util.Objects.requireNonNull;
 
@@ -126,19 +128,12 @@ public SessionHandle openSession(SessionDefinition sessionDefinition) {
    * <p><strong>Error Handling:</strong>
    * If the session does not exist, this method throws an {@link ArmoniKException}
    *
-   * @param sessionId
-   *        the identifier of the existing session to connect to
-   * @param outputListener
-   *        an optional listener for output blob completion events;
-   *        if {@code null}, no listener is registered
-   *
+   * @param sessionId      the identifier of the existing session to connect to
+   * @param outputListener an optional listener for output blob completion events;
+   *                       if {@code null}, no listener is registered
    * @return a {@link SessionHandle} bound to the existing session
-   *
-   * @throws NullPointerException
-   *         if {@code sessionId} is {@code null}
-   * @throws ArmoniKException
-   *         if the session does not exist, or if a communication error occurs
-   *
+   * @throws NullPointerException if {@code sessionId} is {@code null}
+   * @throws ArmoniKException     if the session does not exist, or if a communication error occurs
    * @see SessionHandle
    * @see BlobCompletionListener
    */
@@ -175,22 +170,51 @@ public SessionHandle getSession(SessionId sessionId, BlobCompletionListener outp
    * <p><strong>Error Handling:</strong>
    * If the session does not exist, this method throws an {@link ArmoniKException}.
    *
-   * @param sessionId
-   *        the identifier of the existing session to connect to
-   *
+   * @param sessionId the identifier of the existing session to connect to
    * @return a {@link SessionHandle} for interacting with the session
-   *
-   * @throws NullPointerException
-   *         if {@code sessionId} is {@code null}
-   * @throws ArmoniKException
-   *         if the session does not exist, or if a communication error occurs
-   *
+   * @throws NullPointerException if {@code sessionId} is {@code null}
+   * @throws ArmoniKException     if the session does not exist, or if a communication error occurs
    * @see #getSession(SessionId, BlobCompletionListener)
    */
   public SessionHandle getSession(SessionId sessionId) {
     return getSession(sessionId, null);
   }
 
+  /**
+   * Closes a session in the ArmoniK cluster by its identifier.
+   * <p>
+   * This method sends a request to the Sessions service to transition the session
+   * into the <strong>closed</strong> state. A closed session no longer accepts new task
+   * submissions, but all existing tasks, results, and metadata remain available in the
+   * control plane. Closing a session is the recommended finalization step once no
+   * additional tasks will be submitted.
+   * <p>
+   * This client-level method is provided as a convenience when the caller does not
+   * hold a {@link SessionHandle}. When a {@code SessionHandle} instance is available,
+   * prefer calling {@link SessionHandle#close()} so that all lifecycle-related
+   * operations remain grouped on the handle itself.
+   * <p>
+   * The returned {@link CompletionStage} completes asynchronously with the updated
+   * {@link SessionState} returned by the Sessions service, or completes exceptionally
+   * if the request fails or the session does not exist.
+   *
+   * @param sessionId the identifier of the session to close; must not be {@code null}
+   * @return a completion stage yielding the updated state of the session after the
+   * close operation has been applied
+   * @throws NullPointerException if {@code sessionId} is {@code null}
+   * @see SessionHandle#close()
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> closeSession(SessionId sessionId) {
+    requireNonNull(sessionId, "sessionId must not be null");
+
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.closeSession(toCloseSessionRequest(sessionId)))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+
   /**
    * Closes this client and releases all associated resources.
    * <p>

armonik-client/src/main/java/fr/aneo/armonik/client/SessionHandle.java

@@ -16,6 +16,7 @@
 package fr.aneo.armonik.client;
 
 import fr.aneo.armonik.api.grpc.v1.results.ResultsGrpc;
+import fr.aneo.armonik.api.grpc.v1.sessions.SessionsGrpc;
 import fr.aneo.armonik.client.definition.SessionDefinition;
 import fr.aneo.armonik.client.definition.TaskDefinition;
 import fr.aneo.armonik.client.definition.blob.BlobDefinition;
@@ -29,6 +30,7 @@
 import java.util.function.Function;
 
 import static fr.aneo.armonik.client.internal.grpc.mappers.BlobMapper.toResultMetaDataRequest;
+import static fr.aneo.armonik.client.internal.grpc.mappers.SessionMapper.*;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -187,6 +189,176 @@ public BlobHandle createBlob(InputBlobDefinition blobDefinition) {
     return blobHandle;
   }
 
+  /**
+   * Requests cancellation of this session in the ArmoniK cluster.
+   * <p>
+   * Cancelling a session signals the control plane to stop all remaining work associated
+   * with this session. Depending on the server configuration and current task states,
+   * running tasks may be interrupted and queued tasks will no longer be scheduled.
+   * Results that have already been produced remain accessible unless the session
+   * is subsequently purged or deleted.
+   * <p>
+   * The returned completion stage is completed asynchronously with the updated
+   * {@link SessionState} once the cancellation request has been processed by the
+   * Sessions service, or completed exceptionally if the request fails.
+   *
+   * @return a completion stage that yields the updated state of this session after
+   *         the cancellation request has been applied
+   *
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> cancel() {
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.cancelSession(toCancelSessionRequest(sessionInfo.id())))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+
+  /**
+   * Pauses this session in the ArmoniK cluster.
+   * <p>
+   * Pausing a session temporarily suspends the scheduling of new tasks in the session.
+   * Tasks that are already running continue until completion, but pending tasks are
+   * not picked up for execution until the session is resumed via {@link #resume()}.
+   * <p>
+   * This operation is useful when you need to temporarily throttle or stop processing
+   * without cancelling the session or losing its state.
+   * The returned completion stage is completed asynchronously with the updated
+   * {@link SessionState} once the pause request has been processed by the Sessions
+   * service, or completed exceptionally if the request fails.
+   *
+   * @return a completion stage that yields the updated state of this session after
+   *         the pause request has been applied
+   *
+   * @see #resume()
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> pause() {
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.pauseSession(toPauseSessionRequest(sessionInfo.id())))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+  /**
+   * Resumes a previously paused session in the ArmoniK cluster.
+   * <p>
+   * Resuming a session re-enables scheduling of tasks that were previously held
+   * while the session was paused. Any pending tasks become eligible for execution
+   * again according to the cluster scheduling policies.
+   * <p>
+   * Calling this method on a session that is not paused is safe; the Sessions service
+   * will simply return the current state of the session.
+   * The returned completion stage is completed asynchronously with the updated
+   * {@link SessionState} once the resume request has been processed by the Sessions
+   * service, or completed exceptionally if the request fails.
+   *
+   * @return a completion stage that yields the updated state of this session after
+   *         the resume request has been applied
+   *
+   * @see #pause()
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> resume() {
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.resumeSession(toResumeSessionRequest(sessionInfo.id())))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+
+  /**
+   * Closes this session in the ArmoniK cluster.
+   * <p>
+   * Closing a session finalizes it and prevents any new task submissions, while
+   * preserving existing tasks, results, and metadata. This is the recommended way
+   * to indicate that no further work will be submitted for this session once all
+   * expected tasks have been created.
+   * <p>
+   * Closing a session does not remove its data. To free up storage or completely
+   * remove the session, combine this operation with {@link #purge()} and
+   * {@link #delete()} as appropriate.
+   * The returned completion stage is completed asynchronously with the updated
+   * {@link SessionState} once the close request has been processed by the Sessions
+   * service, or completed exceptionally if the request fails.
+   *
+   * @return a completion stage that yields the updated state of this session after
+   *         the close request has been applied
+   *
+   * @see #purge()
+   * @see #delete()
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> close() {
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.closeSession(toCloseSessionRequest(sessionInfo.id())))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+
+  /**
+   * Purges this session's data in the ArmoniK cluster.
+   * <p>
+   * Purging a session removes the underlying data for its blobs (task inputs and
+   * outputs) from the storage layer while keeping the session and task metadata.
+   * This operation is useful to reclaim storage space once the actual data is no
+   * longer needed but you still want to keep an audit trail or execution history.
+   * <p>
+   * After a purge, attempts to download blob data associated with this session
+   * will fail, but session and task information remain available until the session
+   * is deleted.
+   * The returned completion stage is completed asynchronously with the updated
+   * {@link SessionState} once the purge request has been processed by the Sessions
+   * service, or completed exceptionally if the request fails.
+   *
+   * @return a completion stage that yields the updated state of this session after
+   *         the purge request has been applied
+   *
+   * @see #delete()
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> purge() {
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.purgeSession(toPurgeSessionRequest(sessionInfo.id())))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+
+  /**
+   * Deletes this session from the ArmoniK cluster.
+   * <p>
+   * Deleting a session permanently removes its metadata from the Sessions, Tasks,
+   * and Blobs. This is typically the final step in the lifecycle of a
+   * session once it has been closed and, optionally, purged.
+   * <p>
+   * After deletion, this handle still exists as a local object but any further
+   * interaction with the remote session (such as submitting tasks or querying
+   * state) will fail because the session no longer exists in the cluster.
+   * Clients are expected to discard the handle after deletion.
+   * The returned completion stage is completed asynchronously with the updated
+   * {@link SessionState} reported by the Sessions service just before removal,
+   * or completed exceptionally if the request fails.
+   *
+   * @return a completion stage that yields the last known state of this session
+   *         before it is removed from the cluster
+   *
+   * @see #close()
+   * @see #purge()
+   * @see SessionState
+   */
+  public CompletionStage<SessionState> delete() {
+    return channelPool.executeAsync(channel -> {
+      var sessionsFutureStub = SessionsGrpc.newFutureStub(channel);
+      return Futures.toCompletionStage(sessionsFutureStub.deleteSession(toDeleteSessionRequest(sessionInfo.id())))
+                    .thenApply(response -> toSessionState(response.getSession()));
+    });
+  }
+
+
+
   private Function<ManagedChannel, CompletionStage<BlobInfo>> createBlobInfo(BlobDefinition blobDefinition) {
     return channel -> {
       var request = toResultMetaDataRequest(sessionInfo.id(), List.of(blobDefinition));

armonik-client/src/main/java/fr/aneo/armonik/client/SessionState.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright © 2025 ANEO (armonik@aneo.fr)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package fr.aneo.armonik.client;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * Immutable snapshot of a session's state in the ArmoniK cluster.
+ * <p>
+ * A {@code SessionState} instance mirrors the information returned by the Sessions service
+ * for a given session. It aggregates the current lifecycle status, submission flags,
+ * configuration and key timestamps into a single value object. Instances of this record
+ * are typically obtained from session lifecycle operations such as
+ * {@link SessionHandle#cancel()}, {@link SessionHandle#pause()}, {@link SessionHandle#resume()},
+ * {@link SessionHandle#close()}, {@link SessionHandle#purge()} and {@link SessionHandle#delete()},
+ * or from {@link ArmoniKClient#closeSession(SessionId)}.
+ * <p>
+ * Time-related fields may be {@code null} when the corresponding lifecycle transition
+ * has not occurred yet. For example, {@code cancelledAt} is only populated when the
+ * session has been cancelled, and {@code closedAt} is only populated after the session
+ * has been closed. The {@code duration} field is generally set when the session is in a
+ * terminal state (such as {@code CANCELLED} or {@code CLOSED}) and represents the elapsed
+ * time between creation and termination as computed by the control plane.
+ *
+ * @param sessionId         the unique identifier of the session
+ * @param status            the current lifecycle status of the session as reported by the cluster
+ * @param clientSubmission  whether clients are currently allowed to submit new tasks in this session
+ * @param workerSubmission  whether workers are currently allowed to submit tasks in this session
+ * @param partitionIds      the set of partition identifiers associated with this session; determines where
+ *                          tasks may be scheduled
+ * @param taskConfiguration the default task configuration applied to tasks created in this session
+ * @param createdAt         the instant at which the session was created in the cluster; never {@code null}
+ *                          for a valid session
+ * @param cancelledAt       the instant at which the session was cancelled, or {@code null} if the session
+ *                          has not been cancelled
+ * @param closedAt          the instant at which the session was closed, or {@code null} if the session
+ *                          has not been closed
+ * @param purgedAt          the instant at which the session's data was purged from storage, or {@code null}
+ *                          if no purge has been performed
+ * @param deletedAt         the instant at which the session was deleted from the control plane, or {@code null}
+ *                          if the session still exists
+ * @param duration          the total duration of the session as computed by the control plane, typically set
+ *                          when the session reaches a terminal state; may be {@code null} otherwise
+ * @see SessionHandle
+ * @see SessionStatus
+ * @see TaskConfiguration
+ */
+public record SessionState(
+  SessionId sessionId,
+  SessionStatus status,
+  boolean clientSubmission,
+  boolean workerSubmission,
+  List<String> partitionIds,
+  TaskConfiguration taskConfiguration,
+  Instant createdAt,
+  Instant cancelledAt,
+  Instant closedAt,
+  Instant purgedAt,
+  Instant deletedAt,
+  Duration duration
+) {
+}