Skip to content

refactor(handlers): encode failure in best-effort lookups — 13 silent-success-masking sites (#756 Cat 2) #863

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ClintEastman02 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(handlers): encode failure in best-effort lookups — 13 silent-success-masking sites (#756 Cat 2) #863

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 8 additions & 7 deletions cdk/src/handlers/github-webhook-processor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ import {
findLinearIssueByIdentifier,
} from './shared/linear-issue-lookup';
import { logger } from './shared/logger';
import { type LookupResult, LOOKUP_ABSENT, lookupFailed, lookupFound, lookupValueOr } from './shared/lookup-result';
import { isIntegrationNode } from './shared/orchestration-integration-node';
import { buildScreenshotKey, encodeMarkdownUrl, extractTaskIdFromBranch, isAllowedScreenshotUrl } from './shared/screenshot-url';
import { makeClient, makeDocClient } from './shared/ua';
Expand Down Expand Up @@ -344,7 +345,7 @@ export async function handler(event: ProcessorEvent): Promise<void> {
// link to that reply now (in place). Find the most-recent iteration
// reply id for this issue and edit it; idempotent via the [preview]
// marker so a webhook redelivery won't double-append.
const iter = await findIterationReplyId(linearIssue.issueId, sha);
const iter = lookupValueOr(await findIterationReplyId(linearIssue.issueId, sha), null);
if (iter) {
// (1) Durably persist the screenshot onto the ITERATION task so the
// terminal-settle renders the thumbnail from a strongly-consistent
Expand Down Expand Up @@ -412,8 +413,8 @@ export async function handler(event: ProcessorEvent): Promise<void> {
async function findIterationReplyId(
linearIssueId: string,
deploySha?: string,
): Promise<{ replyId: string; taskId: string } | null> {
if (!TASK_TABLE) return null;
): Promise<LookupResult<{ replyId: string; taskId: string }>> {
if (!TASK_TABLE) return LOOKUP_ABSENT;
try {
const res = await ddb.send(new QueryCommand({
TableName: TASK_TABLE,
Expand All @@ -426,7 +427,7 @@ async function findIterationReplyId(
.map((item) => ({ taskId: item.task_id, replyId: item.channel_metadata?.iteration_reply_comment_id }))
.filter((c): c is { taskId: string; replyId: string } =>
typeof c.taskId === 'string' && typeof c.replyId === 'string' && c.replyId.length > 0);
if (candidates.length === 0) return null;
if (candidates.length === 0) return LOOKUP_ABSENT;

// Prefer the task whose pushed head_sha matches this deploy's commit (correct
// attribution under overlapping iterations). Walk newest-first; GetItem the
Expand All @@ -436,16 +437,16 @@ async function findIterationReplyId(
const got = await ddb.send(new GetCommand({
TableName: TASK_TABLE, Key: { task_id: c.taskId }, ProjectionExpression: 'head_sha',
}));
if (got.Item?.head_sha === deploySha) return { replyId: c.replyId, taskId: c.taskId };
if (got.Item?.head_sha === deploySha) return lookupFound({ replyId: c.replyId, taskId: c.taskId });
}
}
// No SHA match (pre-fix task / non-PR deploy) → newest reply-bearing task.
return { replyId: candidates[0].replyId, taskId: candidates[0].taskId };
return lookupFound({ replyId: candidates[0].replyId, taskId: candidates[0].taskId });
} catch (err) {
logger.warn('findIterationReplyId query failed (non-fatal)', {
linear_issue_id: linearIssueId, error: err instanceof Error ? err.message : String(err),
});
return null;
return lookupFailed(err);
}
}

Expand Down
13 changes: 7 additions & 6 deletions cdk/src/handlers/jira-webhook-processor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ import {
import { resolveSoleActiveJiraTenant } from './shared/jira-tenant-registry';
import type { SubIssueNode } from './shared/linear-subissue-fetch';
import { logger } from './shared/logger';
import { lookupValueOr } from './shared/lookup-result';
import type { CommentRef } from './shared/orchestration-channel';
import { makeJiraChannel } from './shared/orchestration-channel-jira';
import { parseRetryIntent } from './shared/orchestration-comment-trigger';
Expand Down Expand Up @@ -634,7 +635,7 @@ export async function handler(event: ProcessorEvent): Promise<void> {
discovery.orchestrationId,
);
if (fresh) {
const commentId = await upsertEpicPanel({
const commentId = lookupValueOr(await upsertEpicPanel({
channel: makeJiraChannel(WORKSPACE_REGISTRY_TABLE),
parent: {
issueId: issue.key,
Expand All @@ -646,7 +647,7 @@ export async function handler(event: ProcessorEvent): Promise<void> {
},
children: fresh.children,
labelFilter,
});
}), null);
if (commentId) {
await setStatusCommentId(
ddb,
Expand Down Expand Up @@ -718,7 +719,7 @@ export async function handler(event: ProcessorEvent): Promise<void> {
try {
// Unlike seed, extension already has a durable snapshot. Refresh the
// panel from it even if a post-release read is temporarily unavailable.
const commentId = await upsertEpicPanel({
const commentId = lookupValueOr(await upsertEpicPanel({
channel: makeJiraChannel(WORKSPACE_REGISTRY_TABLE),
parent: {
issueId: issue.key,
Expand All @@ -738,7 +739,7 @@ export async function handler(event: ProcessorEvent): Promise<void> {
children: panelSnapshot.children,
inProgress: true,
labelFilter,
});
}), null);
if (commentId && !panelSnapshot.meta.status_comment_id) {
await setStatusCommentId(
ddb,
Expand Down Expand Up @@ -1184,7 +1185,7 @@ async function handleJiraEpicRetry(
if (WORKSPACE_REGISTRY_TABLE) {
const refreshed = await loadOrchestration(ddb, ORCHESTRATION_TABLE, orchestrationId);
if (refreshed) {
const panelId = await upsertEpicPanel({
const panelId = lookupValueOr(await upsertEpicPanel({
channel: makeJiraChannel(WORKSPACE_REGISTRY_TABLE),
parent: {
issueId: parentIssueKey,
Expand All @@ -1202,7 +1203,7 @@ async function handleJiraEpicRetry(
statusCommentId: refreshed.meta.status_comment_id,
inProgress: true,
labelFilter: refreshed.meta.release_context.trigger_label,
});
}), null);
if (panelId) {
await setStatusCommentId(ddb, ORCHESTRATION_TABLE, orchestrationId, panelId);
}
Expand Down
110 changes: 67 additions & 43 deletions cdk/src/handlers/linear-webhook-processor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ import { resolveLinearOauthToken } from './shared/linear-oauth-resolver';
import { fetchIssueParentId } from './shared/linear-subissue-fetch';
import { lookupTaskByLinearIssue, prNumberFromTask } from './shared/linear-task-by-issue';
import { logger } from './shared/logger';
import { type LookupResult, LOOKUP_ABSENT, isLookupFailure, lookupFailed, lookupFound, lookupValueOr } from './shared/lookup-result';
import { type Channel, type IssueRef } from './shared/orchestration-channel';
import { makeLinearChannel } from './shared/orchestration-channel-linear';
import {
Expand All @@ -74,6 +75,7 @@ import { computeEpicRetryPlan } from './shared/orchestration-reconcile';
import { applyTerminalCreateFailures, readConcurrencyBudget, releaseReadyChildren } from './shared/orchestration-release';
import { upsertEpicPanel } from './shared/orchestration-rollup';
import { claimCommentAck, clearRollupClaim, deriveOrchestrationId, loadOrchestration, setChildOwnAttachments, setRetryCommentId, setStatusCommentId, type OrchestrationChildRow, type OrchestrationReleaseContext } from './shared/orchestration-store';
import { readTaskPrNumber } from './shared/task-pr-number';
import { DEFAULT_LABEL_FILTER, hasHelpLabel, HELP_SUFFIX } from './shared/trigger-label';
import type { Attachment, PassedAttachmentRecord } from './shared/types';
import { makeClient, makeDocClient } from './shared/ua';
Expand Down Expand Up @@ -515,7 +517,7 @@ async function postIterationAck(
registryTableName: string,
issueId: string,
replyTargetId: string,
): Promise<string | null> {
): Promise<LookupResult<string>> {
try {
const ref = await channelFor(registryTableName).upsertThreadedReply?.(
issueRef(issueId, workspaceId),
Expand All @@ -524,12 +526,12 @@ async function postIterationAck(
);
// An empty id means the surface posted but can't address the reply later —
// report "no reply to mature" rather than stamping a blank id on the task.
return ref?.commentId || null;
return ref?.commentId ? lookupFound(ref.commentId) : LOOKUP_ABSENT;
} catch (err) {
logger.warn('Iteration ack reply failed (non-fatal)', {
issue_id: issueId, error: err instanceof Error ? err.message : String(err),
});
return null;
return lookupFailed(err);
}
}

Expand Down Expand Up @@ -1173,14 +1175,14 @@ export async function handler(event: ProcessorEvent): Promise<void> {
const settled = seedHadTerminalFailure && postReleaseSnapshot.children.every(
(c) => c.child_status === 'succeeded' || c.child_status === 'failed' || c.child_status === 'skipped',
);
const commentId = await upsertEpicPanel({
const commentId = lookupValueOr(await upsertEpicPanel({
channel: channelFor(WORKSPACE_REGISTRY_TABLE),
parent: issueRef(issue.id, workspaceId),
children: postReleaseSnapshot.children,
...seedFailureReasons(postReleaseSnapshot.children),
inProgress: !settled,
mirrorParentState: true,
});
}), null);
if (commentId) {
await setStatusCommentId(ddb, ORCHESTRATION_TABLE, discovery.orchestrationId, commentId);
}
Expand Down Expand Up @@ -1283,13 +1285,13 @@ export async function handler(event: ProcessorEvent): Promise<void> {
const fresh = await loadOrchestration(ddb, ORCHESTRATION_TABLE, discovery.orchestrationId);
const children = fresh?.children ?? snapshot.children;
const meta = (fresh ?? snapshot).meta;
const newId = await upsertEpicPanel({
const newId = lookupValueOr(await upsertEpicPanel({
channel: channelFor(WORKSPACE_REGISTRY_TABLE),
parent: issueRef(issue.id, workspaceId),
...(meta.status_comment_id !== undefined && { statusCommentId: meta.status_comment_id }),
children,
inProgress: true, // the extend re-opened the epic
});
}), null);
if (newId && meta.status_comment_id === undefined) {
await setStatusCommentId(ddb, ORCHESTRATION_TABLE, discovery.orchestrationId, newId);
}
Expand Down Expand Up @@ -1679,13 +1681,13 @@ async function maybeRetryTerminalEpic(
);
}
// Post the panel FRESH (no statusCommentId → new comment, below the note).
const newPanelId = await upsertEpicPanel({
const newPanelId = lookupValueOr(await upsertEpicPanel({
channel,
parent: parentRef,
children,
inProgress: true,
mirrorParentState: true,
});
}), null);
if (newPanelId) {
await setStatusCommentId(ddb, ORCHESTRATION_TABLE, orchestrationId, newPanelId);
}
Expand Down Expand Up @@ -1929,7 +1931,18 @@ async function handleCommentTrigger(payload: LinearCommentEvent): Promise<void>
// the issue may still be a plain (non-orchestration) issue that ABCA opened
// a PR for — fall through to the standalone path, which iterates
// on that PR with the same 👀/reply ack but no dependency cascade.
const parentId = await fetchIssueParentId(resolved.accessToken, commentedIssueId);
const parentResult = await fetchIssueParentId(resolved.accessToken, commentedIssueId);
if (isLookupFailure(parentResult)) {
// The parent lookup broke (Linear outage / GraphQL error). Do NOT fall
// through to the standalone path — that would silently downgrade an
// orchestration child. Defer; a stream replay re-drives once Linear recovers.
logger.warn('Comment trigger: issue-parent lookup failed — deferring (not downgrading to standalone)', {
issue_id: commentedIssueId,
error: parentResult.error instanceof Error ? parentResult.error.message : String(parentResult.error),
});
return;
}
const parentId = lookupValueOr(parentResult, null);
const orchestrationId = parentId ? deriveOrchestrationId(parentId) : null;
const snapshot = orchestrationId
? await loadOrchestration(ddb, ORCHESTRATION_TABLE, orchestrationId)
Expand Down Expand Up @@ -2083,8 +2096,19 @@ async function handleParentEpicCommentTrigger(args: {
return;
}

const prNumber = await resolveChildPrNumber(target.child_task_id);
if (prNumber === null) {
const prNumberResult = await readTaskPrNumber(ddb, process.env.TASK_TABLE_NAME!, target.child_task_id);
if (isLookupFailure(prNumberResult)) {
// The read broke — don't tell the user "no PR yet" (a distinct, misleading
// state). Leave the 👀 in place so a stream replay can re-drive it.
logger.warn('Comment trigger (parent epic): sub-issue PR read failed — deferring', {
orchestration_id: orchestrationId,
sub_issue_id: target.sub_issue_id,
child_task_id: target.child_task_id,
error: prNumberResult.error instanceof Error ? prNumberResult.error.message : String(prNumberResult.error),
});
return;
}
if (!prNumberResult.ok) {
// Matched a node but it has no PR to iterate. If that node FAILED, the user
// named it to fix it — there's nothing to iterate (no PR), so point them
// straight at retry instead of the generic disambiguation. Observed in
Expand All @@ -2106,6 +2130,7 @@ async function handleParentEpicCommentTrigger(args: {
});
return;
}
const prNumber = prNumberResult.value;

// Resolve the FULL child row (the matcher returns a trimmed view without
// ``repo``) so the iteration carries the sub-issue's repo.
Expand Down Expand Up @@ -2173,12 +2198,32 @@ async function iterateOrchestrationChild(args: {
const subIssueId = child.sub_issue_id;
const triggerCommentIssueId = args.triggerCommentIssueId ?? subIssueId;

const prNumber = args.prNumber ?? (child.child_task_id ? await resolveChildPrNumber(child.child_task_id) : null);
if (prNumber === null || prNumber === undefined) {
logger.warn('Comment trigger: sub-issue has no resolvable PR — cannot iterate', {
orchestration_id: orchestrationId, sub_issue_id: subIssueId, child_task_id: child.child_task_id,
});
return;
let prNumber: number;
if (args.prNumber !== undefined) {
prNumber = args.prNumber;
} else {
const prNumberResult = child.child_task_id
? await readTaskPrNumber(ddb, process.env.TASK_TABLE_NAME!, child.child_task_id)
: LOOKUP_ABSENT;
if (!prNumberResult.ok) {
// Can't iterate without a PR either way, but log an outage distinctly
// from a genuinely-absent PR so it isn't misread as "nothing to do".
logger.warn(
isLookupFailure(prNumberResult)
? 'Comment trigger: sub-issue PR read failed — cannot iterate'
: 'Comment trigger: sub-issue has no resolvable PR — cannot iterate',
{
orchestration_id: orchestrationId,
sub_issue_id: subIssueId,
child_task_id: child.child_task_id,
...(isLookupFailure(prNumberResult) && {
error: prNumberResult.error instanceof Error ? prNumberResult.error.message : String(prNumberResult.error),
}),
},
);
return;
}
prNumber = prNumberResult.value;
}

// Attribute to the orchestration's release user (the comment author may not
Expand All @@ -2198,7 +2243,7 @@ async function iterateOrchestrationChild(args: {
// silence) and persist its id so the fanout dispatcher matures THIS reply
// (🔄→✅/💬) instead of posting new top-level comments. The reply threads under
// the conversation root (replyTargetId) on the issue the comment lives on.
const iterationReplyId = await postIterationAck(workspaceId, registryTableName, triggerCommentIssueId, replyTargetId);
const iterationReplyId = lookupValueOr(await postIterationAck(workspaceId, registryTableName, triggerCommentIssueId, replyTargetId), null);

// Idempotency: one iteration per (sub-issue, comment). The comment id is
// unique per comment, so a webhook retry of the same comment dedups.
Expand Down Expand Up @@ -2433,7 +2478,7 @@ async function handleStandaloneCommentTrigger(args: {
await channel.reactToComment?.({ commentId }, target, 'started');
// Immediate "👀 On it" threaded reply + persist its id so the fanout dispatcher
// matures THIS reply instead of posting new comments.
const iterationReplyId = await postIterationAck(workspaceId, registryTableName, issueId, replyTargetId);
const iterationReplyId = lookupValueOr(await postIterationAck(workspaceId, registryTableName, issueId, replyTargetId), null);

const idempotencyKey = `iterate_${issueId}_${commentId}`.replace(/[^A-Za-z0-9_-]/g, '').slice(0, MAX_IDEMPOTENCY_KEY_LENGTH);
const channelMetadata: Record<string, string> = {
Expand Down Expand Up @@ -2600,7 +2645,7 @@ async function maybeStartStandaloneNewWork(args: {
// ACK immediately (👀 reaction + threaded "On it"), same as the iteration and
// clarify-resume paths.
await channel.reactToComment?.({ commentId }, target, 'started');
const iterationReplyId = await postIterationAck(workspaceId, registryTableName, issueId, replyTargetId);
const iterationReplyId = lookupValueOr(await postIterationAck(workspaceId, registryTableName, issueId, replyTargetId), null);

// Idempotency: key on (issue, comment) so a webhook redelivery of the SAME
// comment doesn't spawn a second task. Distinct prefix from iterate_/clarify_.
Expand Down Expand Up @@ -2729,7 +2774,7 @@ async function maybeResumeClarifyHold(args: {
const channel = channelFor(registryTableName);
const target = issueRef(issueId, workspaceId);
await channel.reactToComment?.({ commentId }, target, 'started');
const iterationReplyId = await postIterationAck(workspaceId, registryTableName, issueId, replyTargetId);
const iterationReplyId = lookupValueOr(await postIterationAck(workspaceId, registryTableName, issueId, replyTargetId), null);

const resumeDescription = buildClarifyResumeDescription(
typeof row.task_description === 'string' ? row.task_description : undefined,
Expand Down Expand Up @@ -2802,27 +2847,6 @@ async function maybeResumeClarifyHold(args: {
return true;
}

/** Read a child task's PR number (numeric pr_number, else parse pr_url). Null if neither. */
async function resolveChildPrNumber(taskId: string): Promise<number | null> {
try {
const res = await ddb.send(new GetCommand({ TableName: process.env.TASK_TABLE_NAME!, Key: { task_id: taskId } }));
const pr = res.Item?.pr_number;
if (typeof pr === 'number') return pr;
const url = res.Item?.pr_url;
if (typeof url === 'string') {
const m = url.match(/\/pull\/(\d+)\b/);
if (m) return Number(m[1]);
}
return null;
} catch (err) {
logger.warn('Comment trigger: failed to read sub-issue task record for PR number', {
task_id: taskId,
error: err instanceof Error ? err.message : String(err),
});
return null;
}
}

/**
* Decide whether a Linear Issue event should trigger a task.
*
Expand Down
Loading
Loading