fix: mark push CANCELED when client disconnects mid-push (#217)

## Summary

- Adds an `onDisconnect` callback to `HeartbeatSender` — fires once when
a sideband write throws, indicating the client has gone away
- `StoreAndForwardReceivePackFactory` wires the callback to
`PushStore.cancel()` on the active push record, using
`validationRecordId` (PENDING) if set, otherwise the initial `pushId`
(RECEIVED)
- No sweeper added — a TTL-based approach cannot distinguish abandoned
RECEIVED records from PENDING records legitimately waiting in the
approval queue

closes #214

## Test plan

- [ ] Push to `/push/` path and Ctrl+C mid-way through the approval
wait; confirm push record transitions to CANCELED
- [ ] Normal push (no disconnect) still reaches FORWARDED correctly
- [ ] Unit tests pass: `./gradlew :git-proxy-java-core:test`

Author: coopernetes

git-proxy-java-core/src/main/java/org/finos/gitproxy/git/HeartbeatSender.java

@@ -15,10 +15,13 @@
  * The dot is harmless whitespace and does not affect validation output. If the interval is zero or negative the sender
  * is a no-op.
  *
+ * <p>An optional {@code onDisconnect} callback is invoked once when a write fails, indicating the client has gone away.
+ * The callback runs on the heartbeat thread and must be short and non-blocking.
+ *
  * <p>Usage:
  *
  * <pre>{@code
- * try (HeartbeatSender hb = new HeartbeatSender(rp, Duration.ofSeconds(10))) {
+ * try (HeartbeatSender hb = new HeartbeatSender(rp, Duration.ofSeconds(10), this::handleDisconnect)) {
  *     hb.start();
  *     // ... long-running hook chain ...
  * }
@@ -34,11 +37,17 @@ public class HeartbeatSender implements AutoCloseable {
 
     private final ReceivePack rp;
     private final Duration interval;
+    private final Runnable onDisconnect;
     private ScheduledExecutorService executor;
 
     public HeartbeatSender(ReceivePack rp, Duration interval) {
+        this(rp, interval, null);
+    }
+
+    public HeartbeatSender(ReceivePack rp, Duration interval, Runnable onDisconnect) {
         this.rp = rp;
         this.interval = interval;
+        this.onDisconnect = onDisconnect;
     }
 
     /** Starts the heartbeat background thread. No-op if interval is zero or negative. */
@@ -61,10 +70,16 @@ private void sendDot() {
             rp.sendMessage(".");
             rp.getMessageOutputStream().flush();
         } catch (Exception e) {
-            // Session may have ended; stop firing
             if (executor != null) {
                 executor.shutdownNow();
             }
+            if (onDisconnect != null) {
+                try {
+                    onDisconnect.run();
+                } catch (Exception ex) {
+                    log.warn("Disconnect callback threw", ex);
+                }
+            }
         }
     }
 

git-proxy-java-core/src/main/java/org/finos/gitproxy/git/StoreAndForwardReceivePackFactory.java

@@ -256,7 +256,25 @@ public ReceivePack create(HttpServletRequest req, Repository db)
             preHooks = validationHooks.toArray(PreReceiveHook[]::new);
         }
 
-        rp.setPreReceiveHook(chainPreReceiveHooks(heartbeatInterval, validationContext, failFast, preHooks));
+        Runnable disconnectCallback = null;
+        if (persistenceHook != null) {
+            final PushContext capturedContext = pushContext;
+            disconnectCallback = () -> {
+                String recordId = capturedContext.getValidationRecordId();
+                if (recordId == null) recordId = capturedContext.getPushId();
+                if (recordId != null) {
+                    try {
+                        pushStore.cancel(recordId, null);
+                        log.info("Push {} marked CANCELED: client disconnected mid-push", recordId);
+                    } catch (Exception e) {
+                        log.warn("Failed to mark push {} as CANCELED after client disconnect", recordId, e);
+                    }
+                }
+            };
+        }
+
+        rp.setPreReceiveHook(
+                chainPreReceiveHooks(heartbeatInterval, validationContext, failFast, disconnectCallback, preHooks));
 
         // Post-receive: forward to upstream, then record final status
         var forwardingHook = new ForwardingPostReceiveHook(provider, creds, pushContext, connectTimeoutSeconds);
@@ -275,9 +293,10 @@ private static PreReceiveHook chainPreReceiveHooks(
             Duration heartbeatInterval,
             ValidationContext validationContext,
             boolean failFast,
+            Runnable disconnectCallback,
             PreReceiveHook... hooks) {
         return (ReceivePack rp, Collection<ReceiveCommand> commands) -> {
-            try (HeartbeatSender heartbeat = new HeartbeatSender(rp, heartbeatInterval)) {
+            try (HeartbeatSender heartbeat = new HeartbeatSender(rp, heartbeatInterval, disconnectCallback)) {
                 heartbeat.start();
                 boolean skipValidationHooks = false;
                 for (PreReceiveHook hook : hooks) {

git-proxy-java-dashboard/src/main/java/org/finos/gitproxy/dashboard/controller/PushController.java

@@ -195,7 +195,7 @@ public record ApproveBody(
             String reviewerEmail,
             String reason,
             Map<String, String> attestations,
-            boolean adminOverride) {}
+            Boolean adminOverride) {}
 
     /**
      * Approve a push. Body: { "reviewerUsername": "...", "reviewerEmail": "...", "reason": "...", "attestations": {
@@ -211,7 +211,8 @@ public ResponseEntity<?> approve(@PathVariable String id, @RequestBody ApproveBo
                         return ResponseEntity.badRequest()
                                 .body(Map.of("error", "Push is not in PENDING status: " + record.getStatus()));
                     }
-                    ResponseEntity<?> identityError = checkReviewerIdentity(record, body.adminOverride());
+                    boolean adminOverride = Boolean.TRUE.equals(body.adminOverride());
+                    ResponseEntity<?> identityError = checkReviewerIdentity(record, adminOverride);
                     if (identityError != null) return identityError;
 
                     // Validate required attestation questions are answered
@@ -225,7 +226,7 @@ public ResponseEntity<?> approve(@PathVariable String id, @RequestBody ApproveBo
                             .reviewerUsername(resolveReviewerFromApproveBody(body, auth))
                             .reviewerEmail(body.reviewerEmail())
                             .reason(body.reason())
-                            .selfApproval(isSelfApproval(record, auth, body.adminOverride()))
+                            .selfApproval(isSelfApproval(record, auth, adminOverride))
                             .answers(body.attestations())
                             .build();
                     var updated = pushStore.approve(id, attestation);

test/proxy-pass.sh

@@ -46,7 +46,7 @@ APPROVE_RESPONSE=$(curl -s -o /dev/null -w "%{http_code}" \
     -X POST "http://localhost:8080/api/push/${PUSH_ID}/authorise" \
     -H "Content-Type: application/json" \
     -H "X-Api-Key: ${GITPROXY_API_KEY}" \
-    -d '{"user":"test-script","comment":"auto-approved by proxy-pass.sh","attestations":{"reviewed-content":"true","policy-compliance":"true"}}')
+    -d '{"reviewerUsername":"test-script","reason":"auto-approved by proxy-pass.sh"}')
 if [ "${APPROVE_RESPONSE}" != "200" ]; then
     echo "ERROR: Approval API returned HTTP ${APPROVE_RESPONSE}"
     exit 1