Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
30 changes: 30 additions & 0 deletions apps/server/src/orchestration/ThreadPullRequestReactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@ import {
} from "@t3tools/contracts";
import { describe, expect, it } from "@effect/vitest";
import * as Crypto from "effect/Crypto";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
Expand Down Expand Up @@ -764,6 +766,34 @@ describe("ThreadPullRequestReactor", () => {
),
);

it.effect("skips the startup backfill for threads older than the backfill window", () =>
Effect.scoped(
Effect.gen(function* () {
const old = DateTime.formatIso(
DateTime.makeUnsafe(
Date.parse(NOW) - Duration.toMillis(ThreadPullRequestReactor.BACKFILL_WINDOW) - 1,
),
);
const fixture = yield* makeHarness({
threads: [
thread("recent", { branch: "recent", settledAt: NOW }),
thread("old", { branch: "old", settledAt: old, updatedAt: old }),
// Auto-settle keeps the last activity as settledAt but stamps updatedAt now.
thread("auto-settled", { branch: "auto-settled", settledAt: old }),
],
branchPullRequest: () => Effect.succeed(branchPullRequest()),
});
yield* TestClock.setTime(Date.parse(NOW));
yield* Effect.gen(function* () {
yield* fixture.start();
expect((yield* Ref.get(fixture.commands)).map((command) => command.threadId)).toEqual([
"recent",
]);
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect("matches Azure SSH projects to HTTPS PRs with the provider repository selector", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down
20 changes: 16 additions & 4 deletions apps/server/src/orchestration/ThreadPullRequestReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,10 @@ import {
} from "@t3tools/contracts";
import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
import * as Cause from "effect/Cause";
import * as Clock from "effect/Clock";
import * as Context from "effect/Context";
import * as Crypto from "effect/Crypto";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
Expand Down Expand Up @@ -54,6 +56,14 @@ function samePullRequest(
/** Startup lookups per settled thread before discovery gives up on it. */
export const BACKFILL_ATTEMPTS = 5;

/**
* Startup discovery only covers threads settled this recently. Use settledAt,
* not updatedAt: auto-settle stamps updatedAt with the settle time, so a mass
* auto-settle would put every old thread back in the window. Older threads
* were checked while active or on an earlier boot.
*/
export const BACKFILL_WINDOW = Duration.days(7);

interface RefreshRequest {
readonly threadId: ThreadId | null;
readonly refresh: boolean;
Expand Down Expand Up @@ -112,9 +122,9 @@ export const make = Effect.gen(function* () {
const repositoryIdentities = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
const crypto = yield* Crypto.Crypto;
const fileSystem = yield* FileSystem.FileSystem;
// Settled threads get one link discovery at startup. Failed lookups retry on
// the periodic pass a few times, then stop until the thread changes or the
// server restarts, so a missing or logged-out CLI cannot loop forever.
// Recently settled threads get one link discovery at startup. Failed lookups
// retry on the periodic pass a few times, then stop until the thread changes
// or the server restarts, so a missing or logged-out CLI cannot loop forever.
const pendingBackfill = new Map<ThreadId, number>();
const finishBackfill = (threads: ReadonlyArray<{ readonly id: ThreadId }>) => {
for (const thread of threads) pendingBackfill.delete(thread.id);
Expand All @@ -138,10 +148,12 @@ export const make = Effect.gen(function* () {
: yield* readSweepSnapshot(snapshots, request.threadId);
const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
if (request.backfill) {
const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(BACKFILL_WINDOW);
for (const thread of snapshot.threads) {
if (
(thread.settledOverride === "settled" || thread.settledAt !== null) &&
thread.branchPullRequest == null
thread.branchPullRequest == null &&
Date.parse(thread.settledAt ?? thread.updatedAt) >= cutoff
) {
pendingBackfill.set(thread.id, BACKFILL_ATTEMPTS);
}
Expand Down
Loading