[Java/C++] Archive variable length recordings.

aeron-agent/src/main/java/io/aeron/agent/ArchiveEventCode.java

@@ -87,7 +87,10 @@ public enum ArchiveEventCode implements EventCode
     REPLAY_SESSION_ERROR(36, -1,
         (event, buffer, offset, builder) -> dissectReplaySessionError(buffer, offset, builder)),
     CATALOG_RESIZE(37, -1,
-        (event, buffer, offset, builder) -> dissectCatalogResize(buffer, offset, builder));
+        (event, buffer, offset, builder) -> dissectCatalogResize(buffer, offset, builder)),
+
+    CMD_IN_PURGE_RECORDING(38, PurgeRecordingRequestDecoder.TEMPLATE_ID,
+        ArchiveEventDissector::dissectControlRequest);
 
     static final int EVENT_CODE_TYPE = EventCodeType.ARCHIVE.getTypeCode();
     private static final ArchiveEventCode[] EVENT_CODE_BY_ID;

aeron-agent/src/main/java/io/aeron/agent/ArchiveEventDissector.java

@@ -80,6 +80,8 @@ final class ArchiveEventDissector
         new TaggedReplicateRequestDecoder();
     private static final StopRecordingByIdentityRequestDecoder STOP_RECORDING_BY_IDENTITY_REQUEST_DECODER =
         new StopRecordingByIdentityRequestDecoder();
+    private static final PurgeRecordingRequestDecoder PURGE_RECORDING_REQUEST_DECODER =
+        new PurgeRecordingRequestDecoder();
     private static final ControlResponseDecoder CONTROL_RESPONSE_DECODER = new ControlResponseDecoder();
 
     private ArchiveEventDissector()
@@ -388,6 +390,15 @@ static void dissectControlRequest(
                 appendStopRecordingByIdentity(builder);
                 break;
 
+            case CMD_IN_PURGE_RECORDING:
+                PURGE_RECORDING_REQUEST_DECODER.wrap(
+                    buffer,
+                    offset + relativeOffset,
+                    HEADER_DECODER.blockLength(),
+                    HEADER_DECODER.version());
+                appendPurgeRecording(builder);
+                break;
+
             default:
                 builder.append(": unknown command");
         }
@@ -785,4 +796,12 @@ private static void appendTaggedReplicate(final StringBuilder builder)
         builder.append(", liveDestination=");
         TAGGED_REPLICATE_REQUEST_DECODER.getLiveDestination(builder);
     }
+
+    private static void appendPurgeRecording(final StringBuilder builder)
+    {
+        builder.append(": controlSessionId=").append(PURGE_RECORDING_REQUEST_DECODER.controlSessionId())
+            .append(", correlationId=").append(PURGE_RECORDING_REQUEST_DECODER.correlationId())
+            .append(", recordingId=").append(PURGE_RECORDING_REQUEST_DECODER.recordingId());
+    }
+
 }

aeron-agent/src/main/java/io/aeron/agent/ArchiveEventEncoder.java

@@ -82,22 +82,14 @@ static void encodeCatalogResize(
         final int offset,
         final int captureLength,
         final int length,
-        final int maxEntries,
         final long catalogLength,
-        final int newMaxEntries,
         final long newCatalogLength)
     {
         int relativeOffset = encodeLogHeader(encodingBuffer, offset, captureLength, length);
 
-        encodingBuffer.putInt(offset + relativeOffset, maxEntries, LITTLE_ENDIAN);
-        relativeOffset += SIZE_OF_INT;
-
         encodingBuffer.putLong(offset + relativeOffset, catalogLength, LITTLE_ENDIAN);
         relativeOffset += SIZE_OF_LONG;
 
-        encodingBuffer.putInt(offset + relativeOffset, newMaxEntries, LITTLE_ENDIAN);
-        relativeOffset += SIZE_OF_INT;
-
         encodingBuffer.putLong(offset + relativeOffset, newCatalogLength, LITTLE_ENDIAN);
     }
 

aeron-agent/src/main/java/io/aeron/agent/ArchiveEventLogger.java

@@ -135,10 +135,9 @@ public void logReplaySessionError(final long sessionId, final long recordingId,
         }
     }
 
-    public void logCatalogResize(
-        final int maxEntries, final long catalogLength, final int newMaxEntries, final long newCatalogLength)
+    public void logCatalogResize(final long catalogLength, final long newCatalogLength)
     {
-        final int length = SIZE_OF_LONG * 2 + 2 * SIZE_OF_INT;
+        final int length = SIZE_OF_LONG * 2;
         final int captureLength = captureLength(length);
         final int encodedLength = encodedLength(captureLength);
         final ManyToOneRingBuffer ringBuffer = this.ringBuffer;
@@ -153,9 +152,7 @@ public void logCatalogResize(
                     index,
                     captureLength,
                     length,
-                    maxEntries,
                     catalogLength,
-                    newMaxEntries,
                     newCatalogLength);
             }
             finally

aeron-agent/src/main/java/io/aeron/agent/ArchiveInterceptor.java

@@ -56,10 +56,9 @@ static void onPendingError(final long sessionId, final long recordingId, final S
     static class Catalog
     {
         @Advice.OnMethodEnter
-        static void catalogResized(
-            final int maxEntries, final long catalogLength, final int newMaxEntries, final long newCatalogLength)
+        static void catalogResized(final long catalogLength, final long newCatalogLength)
         {
-            LOGGER.logCatalogResize(maxEntries, catalogLength, newMaxEntries, newCatalogLength);
+            LOGGER.logCatalogResize(catalogLength, newCatalogLength);
         }
     }
 }

aeron-agent/src/test/java/io/aeron/agent/ArchiveEventDissectorTest.java

@@ -833,4 +833,24 @@ void catalogResize()
             " 24 entries (100 bytes) => 777 entries (10000000000 bytes)",
             builder.toString());
     }
+
+    @Test
+    void controlRequestPurgeRecording()
+    {
+        internalEncodeLogHeader(buffer, 0, 56, 901, () -> 1_125_000_000L);
+        final PurgeRecordingRequestEncoder requestEncoder = new PurgeRecordingRequestEncoder();
+        requestEncoder.wrapAndApplyHeader(buffer, LOG_HEADER_LENGTH, headerEncoder)
+            .controlSessionId(15)
+            .correlationId(421)
+            .recordingId(6);
+
+        dissectControlRequest(CMD_IN_PURGE_RECORDING, buffer, 0, builder);
+
+        assertEquals("[1.125] " + CONTEXT + ": " + CMD_IN_PURGE_RECORDING.name() + " [56/901]:" +
+            " controlSessionId=15" +
+            ", correlationId=421" +
+            ", recordingId=6",
+            builder.toString());
+    }
+
 }

aeron-agent/src/test/java/io/aeron/agent/ArchiveEventEncoderTest.java

@@ -94,20 +94,16 @@ void testEncodeCatalogResize()
         final int offset = 24;
         final int length = SIZE_OF_LONG * 2 + SIZE_OF_INT * 2;
         final int captureLength = captureLength(length);
-        final int maxEntries = 3;
         final long catalogLength = 128;
-        final int newMaxEntries = 7;
         final long newCatalogLength = 1024;
 
         encodeCatalogResize(
-            buffer, offset, captureLength, length, maxEntries, catalogLength, newMaxEntries, newCatalogLength);
+            buffer, offset, captureLength, length, catalogLength, newCatalogLength);
 
         assertEquals(captureLength, buffer.getInt(offset, LITTLE_ENDIAN));
         assertEquals(length, buffer.getInt(offset + SIZE_OF_INT, LITTLE_ENDIAN));
         assertNotEquals(0, buffer.getLong(offset + SIZE_OF_INT * 2, LITTLE_ENDIAN));
-        assertEquals(maxEntries, buffer.getInt(offset + LOG_HEADER_LENGTH));
-        assertEquals(catalogLength, buffer.getLong(offset + LOG_HEADER_LENGTH + SIZE_OF_INT));
-        assertEquals(newMaxEntries, buffer.getInt(offset + LOG_HEADER_LENGTH + SIZE_OF_INT + SIZE_OF_LONG));
-        assertEquals(newCatalogLength, buffer.getLong(offset + LOG_HEADER_LENGTH + 2 * SIZE_OF_INT + SIZE_OF_LONG));
+        assertEquals(catalogLength, buffer.getLong(offset + LOG_HEADER_LENGTH));
+        assertEquals(newCatalogLength, buffer.getLong(offset + LOG_HEADER_LENGTH + SIZE_OF_LONG));
     }
 }

aeron-agent/src/test/java/io/aeron/agent/ArchiveEventLoggerTest.java

@@ -189,21 +189,16 @@ void logCatalogResize()
     {
         final int offset = ALIGNMENT * 3;
         logBuffer.putLong(CAPACITY + TAIL_POSITION_OFFSET, offset);
-        final int captureLength = SIZE_OF_LONG * 2 + SIZE_OF_INT * 2;
-        final int maxEntries = 21;
+        final int captureLength = SIZE_OF_LONG * 2;
         final long catalogLength = 42;
-        final int newMaxEntries = 121;
         final long newCatalogLength = 142;
 
-        logger.logCatalogResize(maxEntries, catalogLength, newMaxEntries, newCatalogLength);
+        logger.logCatalogResize(catalogLength, newCatalogLength);
 
         verifyLogHeader(logBuffer, offset, toEventCodeId(CATALOG_RESIZE), captureLength, captureLength);
-        assertEquals(maxEntries, logBuffer.getInt(encodedMsgOffset(offset + LOG_HEADER_LENGTH), LITTLE_ENDIAN));
         assertEquals(catalogLength,
-            logBuffer.getLong(encodedMsgOffset(offset + LOG_HEADER_LENGTH + SIZE_OF_INT), LITTLE_ENDIAN));
-        assertEquals(newMaxEntries, logBuffer.getInt(
-            encodedMsgOffset(offset + LOG_HEADER_LENGTH + SIZE_OF_INT + SIZE_OF_LONG), LITTLE_ENDIAN));
-        assertEquals(newCatalogLength, logBuffer.getLong(
-            encodedMsgOffset(offset + LOG_HEADER_LENGTH + SIZE_OF_INT * 2 + SIZE_OF_LONG), LITTLE_ENDIAN));
+            logBuffer.getLong(encodedMsgOffset(offset + LOG_HEADER_LENGTH), LITTLE_ENDIAN));
+        assertEquals(newCatalogLength,
+            logBuffer.getLong(encodedMsgOffset(offset + LOG_HEADER_LENGTH + SIZE_OF_LONG), LITTLE_ENDIAN));
     }
 }

aeron-archive/src/main/cpp/CMakeLists.txt

@@ -36,6 +36,7 @@ set(GENERATED_CODECS
     ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/ExtendRecordingRequest2.h
     ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/RecordingPositionRequest.h
     ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/TruncateRecordingRequest.h
+    ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/PurgeRecordingRequest.h
     ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/StopRecordingSubscriptionRequest.h
     ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/StartPositionRequest.h
     ${ARCHIVE_CODEC_TARGET_DIR}/aeron_archive_client/StopPositionRequest.h

aeron-archive/src/main/cpp/client/AeronArchive.h

@@ -1205,6 +1205,30 @@ class AeronArchive
         pollForResponse<IdleStrategy>(m_lastCorrelationId);
     }
 
+    /**
+     * Purge a stopped recording, i.e. mark recording as 'RecordingState#INVALID' and delete the corresponding segment
+     * files. The space in the Catalog will be reclaimed upon compaction.
+     *
+     * @param recordingId      of the stopped recording to be purged.
+     * @tparam IdleStrategy to use for polling operations.
+     */
+    template<typename IdleStrategy = aeron::concurrent::BackoffIdleStrategy>
+    inline void purgeRecording(std::int64_t recordingId)
+    {
+        std::lock_guard<std::recursive_mutex> lock(m_lock);
+        ensureOpen();
+        ensureNotReentrant();
+
+        m_lastCorrelationId = m_aeron->nextCorrelationId();
+
+        if (!m_archiveProxy->purgeRecording<IdleStrategy>(recordingId, m_lastCorrelationId, m_controlSessionId))
+        {
+            throw ArchiveException("failed to send purge recording request", SOURCEINFO);
+        }
+
+        pollForResponse<IdleStrategy>(m_lastCorrelationId);
+    }
+
     /**
      * List active recording subscriptions in the archive. These are the result of requesting one of
      * #startRecording(String, int, SourceLocation) or a

aeron-archive/src/main/cpp/client/ArchiveProxy.cpp

@@ -50,6 +50,7 @@
 #include "aeron_archive_client/KeepAliveRequest.h"
 #include "aeron_archive_client/ChallengeResponse.h"
 #include "aeron_archive_client/TaggedReplicateRequest.h"
+#include "aeron_archive_client/PurgeRecordingRequest.h"
 
 using namespace aeron;
 using namespace aeron::concurrent;
@@ -461,6 +462,22 @@ util::index_t ArchiveProxy::truncateRecording(
     return messageAndHeaderLength(request);
 }
 
+util::index_t ArchiveProxy::purgeRecording(
+        AtomicBuffer &buffer,
+        std::int64_t recordingId,
+        std::int64_t correlationId,
+        std::int64_t controlSessionId)
+{
+    PurgeRecordingRequest request;
+
+    wrapAndApplyHeader(request, buffer)
+            .controlSessionId(controlSessionId)
+            .correlationId(correlationId)
+            .recordingId(recordingId);
+
+    return messageAndHeaderLength(request);
+}
+
 util::index_t ArchiveProxy::listRecordingSubscriptions(
     AtomicBuffer &buffer,
     std::int32_t pseudoIndex,

aeron-archive/src/main/cpp/client/ArchiveProxy.h

@@ -574,6 +574,26 @@ class ArchiveProxy
         return offer<IdleStrategy>(m_buffer, 0, length);
     }
 
+    /**
+     * Purge a stopped recording, i.e. mark recording as 'RecordingState#INVALID' and delete the corresponding segment
+     * files. The space in the Catalog will be reclaimed upon compaction.
+     *
+     * @param recordingId      of the stopped recording to be purged.
+     * @param correlationId    for this request.
+     * @param controlSessionId for this request.
+     * @return true if successfully offered otherwise false.
+     */
+    template<typename IdleStrategy = aeron::concurrent::BackoffIdleStrategy>
+    bool purgeRecording(
+            std::int64_t recordingId,
+            std::int64_t correlationId,
+            std::int64_t controlSessionId)
+    {
+        const util::index_t length = purgeRecording(m_buffer, recordingId, correlationId, controlSessionId);
+
+        return offer<IdleStrategy>(m_buffer, 0, length);
+    }
+
     /**
      * List registered subscriptions in the archive which have been used to record streams.
      *
@@ -1042,6 +1062,12 @@ class ArchiveProxy
         std::int64_t correlationId,
         std::int64_t controlSessionId);
 
+    static util::index_t purgeRecording(
+        AtomicBuffer &buffer,
+        std::int64_t recordingId,
+        std::int64_t correlationId,
+        std::int64_t controlSessionId);
+
     static util::index_t listRecordingSubscriptions(
         AtomicBuffer &buffer,
         std::int32_t pseudoIndex,