fix: deduplicate heartbeat checks by queue task by AchoArnold · Pull Request #1080 · NdoleStudio/httpsms

AchoArnold · 2026-11-07T12:24:46Z

Summary

Ignore heartbeat checks without a delivery task ID before database access or recovery scheduling.
Deduplicate heartbeat check streams by comparing the delivered queue task's full resource name with the monitor's stored queue_id; log and ignore mismatches before reading heartbeat history, notifying phones, or scheduling another check.
Reconstruct the task resource name from X-CloudTasks-TaskName and the configured queue, and carry it in typed context metadata under queue.task.id (CloudEvents disallows dotted extension names). Keep emulator and local timeout-fallback delivery compatible.
Log listener failures and acknowledge valid events without requesting HTTP retries, avoiding unsafe replays of subscribers without deduplication.
On temporary monitor-load failures, attempt to schedule a successor check. Log heartbeat-history read failures and stop; log failed queue-ID saves and fail open.
Recovery is best-effort: later heartbeats restart checks only after the existing two-hour inactivity threshold. If the phone stays offline, a failed queue-ID save can leave monitoring stopped; orphaned tasks are discarded by the stored-ID guard.
The task-ID comparison suppresses stale streams but is not an atomic processing claim; overlapping deliveries can still duplicate side effects or supersede a concurrent restart.
Leave queue_id unchanged when heartbeat history no longer exists; stop without a successor and preserve timestamp-based restarts for inactive monitors.

Validation

go test ./... -timeout 120s (API): passed, including the previously failing heartbeat-history restart regression.
API build and go vet for changed packages: passed.
gofumpt, goimports, and golint for changed files: passed (Go tools invoked directly because the pre-commit shell wrappers do not resolve Windows paths).
Regression coverage includes current/stale/missing task IDs, redelivery, database errors, emulator and local fallback delivery, and restarting stopped checks.

Match delivered task IDs against each monitor's queue_id before processing. Preserve emulator and local fallback delivery. Restart stopped checks when heartbeat history resumes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

codacy-production · 2026-11-07T12:26:38Z

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

View in Codacy

🟢 Metrics 39 complexity · 18 duplication

Metric Results

Complexity 39

Duplication 18

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer}
_{TIP This summary will be updated as you push new changes.}

greptile-apps · 2026-11-07T12:27:01Z

Confidence Score: 2/5

[Medium impact] The changes affect the heartbeat-monitoring workflow and its queued checks.

This PR needs fixes before merging because database errors and an overlapping restart can stop heartbeat checks.

Findings

Database errors stop checks ▶
Unsaved task IDs stop checks ▶
Old checks erase replacements ▶

Summary

This PR rejects heartbeat checks whose delivered task ID differs from the monitor's saved QueueID.

Heartbeat checks continue only when their task matches the monitor.
A new heartbeat restarts a stopped monitor check.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart TD
  A[Delivered heartbeat check] --> B[Load monitor]
  B -->|Read fails| C[Return listener error]
  C --> D[Error logged; HTTP returns 204]
  B -->|Read succeeds| E{Task ID matches saved ID?}
  E -->|No| F[Ignore check]
  E -->|Yes| G[Read heartbeat history]
  G -->|Missing| H[Clear saved task ID]
  G -->|Present| I[Check phone and queue next task]
  I --> J[Save next task ID]
  J -->|Write fails| K[Next task is rejected]

_{Reviews (1) · Last reviewed commit: "fix: deduplicate heartbeat checks by que..." · Reviewed by Greptile}

greptile-apps · 2026-11-07T12:27:04Z

 	if err != nil {
-		ctxLogger.Error(stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))
-		return service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), params)
+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))


Database errors stop checks

A temporary monitor-load failure now stops the check stream. Monitor returns an error, but Publish only logs listener errors, so DispatchSync still succeeds and /v1/events returns 204. The queue receives a success response without another check being scheduled. Monitoring stays stopped until a later heartbeat or phone update meets RequiresCheck().

Forward this failure to the HTTP response or retain a safe retry path.

Fixed in 291aaca. Publish now collects listener errors and DispatchSync propagates them for heartbeat checks, so /v1/events returns 500 and Cloud Tasks retries the same task after monitor/history read failures. Other event types retain their existing acknowledgement behavior. Added HTTP failure-then-success coverage and a heartbeat-read failure regression.

Temporary load failure should recover by scheduling a new check itself and if not it's still fine, the monitor will recover when a new heartbeat is received.

Self-scheduling on a monitor-load failure is a reasonable recovery path. One caveat: a new heartbeat does not always restart the check immediately—Store only schedules when monitor.RequiresCheck() is true, and that condition waits until UpdatedAt is more than two hours old (api/pkg/entities/heartbeat_monitor.go).

With the current error return and the propagation added in 291aacaa, the same Cloud Task is retried rather than acknowledged without a successor. If the intended fallback is to let monitoring remain stopped until the two-hour threshold or a monitor update, that’s a valid tradeoff, but it won’t necessarily recover on the next heartbeat.

Updated in 832853b: a temporary monitor-load failure is logged and schedules a successor check itself. Missing monitors still stop, and heartbeat-history read failures are logged and stop without replaying the HTTP event. Heartbeat-based recovery retains the existing two-hour RequiresCheck threshold, so it is best-effort and not necessarily triggered by the very next heartbeat. This limitation is now explicit in the PR description.

greptile-apps · 2026-11-07T12:27:04Z

+	if params.QueueID != monitor.QueueID {
+		ctxLogger.Info(fmt.Sprintf("ignoring heartbeat check with queue ID [%s] for monitor [%s] with current queue ID [%s]", params.QueueID, monitor.ID, monitor.QueueID))
+		return nil


Unsaved task IDs stop checks

The new comparison makes a failed UpdateQueueID write stop monitoring. scheduleHeartbeatCheck queues the next task first, then logs and ignores any error saving its ID. If that write fails, the next task arrives with its new ID while the monitor still holds the previous ID. This comparison discards the valid task as stale, leaving no further checks.

Make saving the next task's ID a required, recoverable part of scheduling before relying on it to reject deliveries.

Fixed in 291aaca. scheduleHeartbeatCheck now returns an error when saving the next queue ID fails. That error reaches the heartbeat consumer as HTTP 500, leaving the original saved ID retryable; retrying the original check can persist a replacement. Added a regression covering a failed save, successful retry, rejection of the orphan task, and continued processing of the valid replacement.

The check will be resumed eventaully by new hearatbeats instead of complicating with the 500 response, let's fail open heartbeat checks queue updates.

Understood—the fail-open approach avoids returning HTTP 500 for a queue-ID persistence failure. One caveat: recovery via a later heartbeat is not guaranteed; if the phone stays offline, there may be no heartbeat to re-arm monitoring, and the orphaned task will be ignored because its ID does not match the stored queue ID. If that gap in offline detection is an accepted tradeoff, fail-open is consistent with that recovery model.

Changed queue-ID saves to fail open in 832853b: persistence failures are logged and do not cause an HTTP 500. An orphaned successor is rejected by the stored-ID guard, and an eligible later heartbeat can establish a replacement stream. Recovery is not guaranteed if the phone remains offline, and the existing two-hour restart threshold still applies. The PR description documents that best-effort tradeoff, and the regression verifies recovery through a later heartbeat rather than HTTP replay.

greptile-apps · 2026-11-07T12:27:04Z


 	heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner)
 	if stacktrace.GetCode(err) == repositories.ErrCodeNotFound {
+		if err = service.monitorRepository.UpdateQueueID(ctx, monitor.ID, ""); err != nil {


Old checks erase replacements

Clearing QueueID here can erase a newer check started by a heartbeat. A check delayed for more than two hours can pass the ID comparison and find no history. Before this clear runs, an incoming heartbeat can see the old UpdatedAt, schedule a replacement, and save its new ID. This unconditional clear then erases that ID, so the replacement is discarded by the new comparison and monitoring stops.

Clear only the task ID this check observed, so an older check cannot cancel its replacement.

Fixed in 291aaca. Added ClearQueueID with an atomic monitor-ID + observed-queue-ID predicate in both GORM and MongoDB repositories. Monitor passes the incoming check's queue ID, so a newer replacement is not cleared. Added deterministic overlap coverage showing that a heartbeat can restart the monitor while an older check reads missing history, plus a regression for the SQL update predicate.

The conditional-clear fix was superseded in 83e5783 following the maintainer's review: checks with no heartbeat history now leave queue_id untouched and return without a successor. With no clearing write, an old check cannot erase a replacement task. Timestamp-based restart behavior is preserved, and the overlapping-replacement regression still passes.

Return heartbeat listener errors to queue consumers for retry. Require next-task ID persistence and conditionally clear observed task IDs. Cover HTTP retries, failed writes, and overlapping heartbeat restarts. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

AchoArnold · 2026-11-07T12:37:31Z


 	heartbeat, err := service.repository.Last(ctx, params.UserID, params.Owner)
 	if stacktrace.GetCode(err) == repositories.ErrCodeNotFound {
+		if err = service.monitorRepository.UpdateQueueID(ctx, monitor.ID, ""); err != nil {


Why are we clearing the qeuueID? Why is this needed?

Clearing is not needed for deduplication itself; it fixes the stopped-monitor restart behavior from the failing test you asked me to fix. When heartbeat history has expired, this check returns without scheduling a successor. Its stored queue ID would then point to a consumed task. Clearing it marks the stream inactive, and RequiresCheck() treats an empty ID as a reason to schedule a check immediately when a new heartbeat arrives. Otherwise a recent UpdatedAt can delay restarting for up to two hours. The clear is conditional on the incoming task ID, so it cannot erase a concurrently scheduled replacement.

Remove this logic. Maybe fix the test or remove the test if if it's not rerelevant.

WHen heartbeat has expired this check shoudl return without a scucesor that's valid logic.

When a new check arrives, RequiresCheck should become true since it hasb een more than 60 days. THe heartbeats are deleted after 60 days

Removed in 83e5783. Missing heartbeat history now stops the check without a successor or any queue_id mutation. RequiresCheck is timestamp-based again. The restart test now models a long-inactive monitor (61 days) rather than requiring an immediate restart from a recently updated monitor, and the clearing helpers/tests were removed. Full API tests, build, vet and lint passed.

Carry queue task IDs in a typed context key named queue.task.id. CloudEvents rejects dotted extension names, so keep this metadata internal. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

Stop checks without clearing task IDs when heartbeat history is absent. Restore timestamp-based RequiresCheck behavior and remove clearing helpers. Model long-inactive monitors in retention restart tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

Separate monitor lookup and deduplication from heartbeat timing decisions. Calculate heartbeat age once and reuse the missed-heartbeat window. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

AchoArnold · 2026-11-07T12:54:24Z

 		return h.responseForbidden(c)
 	}

+	queueID := ""


Put this logic to get/set queue ID in a seprate private method.

Fixed in 832853b: the handler now delegates task-name normalization and typed context assignment to the private withQueueTaskID method. Existing delivery tests cover both a populated task header and a missing header.

AchoArnold · 2026-11-07T12:57:44Z

+	if taskName := c.Get("X-CloudTasks-TaskName"); taskName != "" {
+		queueID = fmt.Sprintf("%s/tasks/%s", h.queueConfig.Name, taskName)
+	}
+	ctx = context.WithValue(ctx, events.ExtensionQueueID, queueID)


Why did you use context instead of cloudevents.extension.

The requested queue.task.id name is not a valid CloudEvent extension name: the Go SDK rejects the dots. This is delivery-task metadata rather than event data, so typed request context preserves that name without invalidating the CloudEvent. HTTP delivery and the local fallback both populate the same key for the listener.

AchoArnold · 2026-11-07T12:58:51Z

 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
 	}

+	queueID, _ := ctx.Value(events.ExtensionQueueID).(string)


if we're using ctx maybe we should name the value QuqueIdKey or perhpas use something from opentelemetry semconv?

Renamed it to QueueIDKey in 832853b, retaining the private typed context key and the exact queue.task.id value. OpenTelemetry semantic-convention keys describe telemetry attributes; they do not provide the context-value contract needed to pass the delivery task ID between the handler and listener.

AchoArnold · 2026-11-07T12:59:44Z

 		),
 	)
+	if len(failures) != 0 {
+		return dispatcher.tracer.WrapErrorSpan(span, stacktrace.Propagatef(errors.Join(failures...), "cannot publish event [%s] with ID [%s]", event.Type(), event.ID()))


Will this cause events to be replayed if one of the handlers fail? Replaying events is not safe since all handlers don't have deduplication logic. Rather use logging in the services to log whhen there's an issue so we can manually fix.

Removed the heartbeat-specific listener-error propagation and Publish error aggregation in 832853b. Listener failures are logged, and valid events are acknowledged with HTTP 204 rather than replaying every subscriber. The regression now verifies that a failing heartbeat listener and a failing ordinary listener are both acknowledged.

AchoArnold · 2026-11-07T13:01:43Z

 	if err != nil {
-		ctxLogger.Error(stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))
-		return service.scheduleHeartbeatCheck(ctx, time.Now().UTC(), params)
+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))


Temporary load failure should recover by scheduling a new check itself and if not it's still fine, the monitor will recover when a new heartbeat is received.

AchoArnold · 2026-11-07T13:02:38Z

+	if params.QueueID != monitor.QueueID {
+		ctxLogger.Info(fmt.Sprintf("ignoring heartbeat check with queue ID [%s] for monitor [%s] with current queue ID [%s]", params.QueueID, monitor.ID, monitor.QueueID))
+		return nil


The check will be resumed eventaully by new hearatbeats instead of complicating with the 500 response, let's fail open heartbeat checks queue updates.

Extract task context handling and rename the typed queue task key. Log heartbeat lookup and queue-ID save errors with best-effort recovery. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

Copilot

🟡 Changes recommended

Queue ownership advances before offline dispatch succeeds and lacks atomic claiming, allowing failed work to lose retries, overlapping deliveries to send duplicate notifications, and phones to remain incorrectly marked online.

4 open issues

Atomically claim task ownership before processing · New
Persist offline action before advancing queue ownership · New
Reject deliveries with an empty task ID · New
Recover when initial queue ID persistence fails · New

What changed in this PR

This PR adds queue-task identity checks to httpSMS heartbeat monitoring to suppress stale check streams and improve failure handling.

Changes:

Pass delivered task IDs through typed context metadata, including emulator and local fallback delivery.
Reject mismatched heartbeat checks and propagate heartbeat-processing failures for HTTP retries.
Add regression coverage for task identity, redelivery, scheduling failures, and inactive-monitor restarts.

File	Description
`api/pkg/services/heartbeat_service.go`	Validates task identity and propagates monitoring and scheduling errors.
`api/pkg/services/heartbeat_service_test.go`	Tests scheduling failures, restarts, and concurrent replacement preservation.
`api/pkg/services/event_dispatcher_service.go`	Returns listener errors and supplies fallback task identity.
`api/pkg/services/event_dispatcher_service_test.go`	Tests fallback and emulator task identity.
`api/pkg/services/emulator_push_queue.go`	Generates full task resource names and sends task headers.
`api/pkg/listeners/heartbeat_listener.go`	Passes task identity into heartbeat monitoring.
`api/pkg/listeners/heartbeat_listener_test.go`	Tests stale-task rejection, redelivery, and load failures.
`api/pkg/handlers/events_handler.go`	Reconstructs task resource names from delivery headers.
`api/pkg/handlers/events_handler_test.go`	Tests task metadata and heartbeat-specific retry responses.
`api/pkg/events/listeners.go`	Defines the typed context key for queue-task identity.

_{Review effort: Balanced}

AchoArnold · 2026-11-07T13:20:33Z

+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))
+	}
+
+	if params.QueueID != monitor.QueueID {


The snapshot check does not guarantee exactly-once side effects for overlapping deliveries, and a concurrent restart can lose ownership to an in-flight successor update. This PR implements the requested eventual suppression of stale check streams, not an atomic claim/lease protocol. A stronger ownership protocol needs a separate recovery design; the owner explicitly rejected HTTP replay and requested simpler fail-open behavior in the preceding review. I have documented this concurrency limitation rather than introducing claims or retry semantics beyond that scope.

AchoArnold · 2026-11-07T13:20:33Z

-		time.Now().UTC().Sub(heartbeat.Timestamp) < (heartbeatCheckInterval*5) && monitor.PhoneOnline {
-		return service.handleFailedMonitor(ctx, heartbeat.Timestamp, params)
+		if heartbeatAge > heartbeatCheckInterval*4 && monitor.PhoneOnline {
+			return service.handleFailedMonitor(ctx, heartbeat.Timestamp, params)


This review targets 724b209. In 832853b, valid events are acknowledged with HTTP 204 even when a listener fails; an offline dispatch failure is logged, not returned as HTTP 500 for replay. The owner explicitly requested that behavior because replaying every subscriber is unsafe. The possibility of losing an offline action after dispatch failure remains a best-effort limitation; durable outbox/retry handling would require a separate design and is not being added to this task-ID filtering change.

AchoArnold · 2026-11-07T13:20:33Z

+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot check if monitor exists with userID [%s] and owner [%s]", params.UserID, params.Owner))
+	}
+
+	if params.QueueID != monitor.QueueID {


Fixed in 13c10dc: a missing delivery task ID is now logged and ignored before loading the monitor. This also prevents the monitor-load recovery path from scheduling an anonymous stream. Added regressions for both IDs empty and for a missing delivery ID while the monitor repository is unavailable; full API tests, build, vet and lint pass.

AchoArnold · 2026-11-07T13:20:33Z

 	if err = service.monitorRepository.UpdateQueueID(ctx, params.MonitorID, queueID); err != nil {
-		service.logger.Error(stacktrace.Propagatef(err, "cannot update monitor with id [%s] with queue with ID [%s]", params.MonitorID, queueID))
+		return service.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot update monitor with id [%s] with queue with ID [%s]", params.MonitorID, queueID))


The delayed recovery is intentional under the latest owner review: failed queue-ID saves are logged and acknowledged, RequiresCheck remains timestamp-based, and empty IDs must not trigger immediate restart or clearing. The existing two-hour threshold also applies to incomplete initial scheduling. This best-effort recovery tradeoff is documented in the PR description and discussed in the replies to 4501827635 and 4501830015; I am not restoring the empty-ID restart condition that the owner previously removed.

Stop anonymous checks before monitor lookup or recovery scheduling. Cover empty stored IDs and unavailable monitors in listener regressions. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 742e0212-0866-4232-89d8-7d194134448a

greptile-apps Bot reviewed Nov 7, 2026

View reviewed changes

AchoArnold commented Nov 7, 2026

View reviewed changes

AchoArnold and others added 3 commits November 7, 2026 14:42

AchoArnold commented Nov 7, 2026

View reviewed changes

AchoArnold requested a balanced review from Copilot November 7, 2026 13:07

Copilot started reviewing on behalf of AchoArnold November 7, 2026 13:07

View session

Copilot AI reviewed Nov 7, 2026

View reviewed changes

Conversation

AchoArnold commented Nov 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Validation

Uh oh!

codacy-production Bot commented Nov 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Up to standards ✅

Uh oh!

greptile-apps Bot commented Nov 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Confidence Score: 2/5

Findings

Summary

Diagram

Uh oh!

greptile-apps Bot Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

AchoArnold commented Nov 7, 2026 •

edited

Loading

codacy-production Bot commented Nov 7, 2026 •

edited

Loading

greptile-apps Bot commented Nov 7, 2026 •

edited

Loading