| ... | @@ -167,6 +167,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: | ... | @@ -167,6 +167,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: |
| 167 | const snapshotReadCount = yield* Ref.make(0); | 167 | const snapshotReadCount = yield* Ref.make(0); |
| 168 | const snapshotReads = yield* Queue.unbounded<number>(); | 168 | const snapshotReads = yield* Queue.unbounded<number>(); |
| 169 | const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS); | 169 | const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS); |
| | 170 | const settingsReads = yield* Queue.unbounded<ServerSettings>(); |
| 170 | const settingsChanges = yield* PubSub.unbounded<ServerSettings>(); | 171 | const settingsChanges = yield* PubSub.unbounded<ServerSettings>(); |
| 171 | const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>(); | 172 | const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>(); |
| 172 | const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]); | 173 | const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]); |
| ... | @@ -226,7 +227,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: | ... | @@ -226,7 +227,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: |
| 226 | const serverSettings = ServerSettingsService.of({ | 227 | const serverSettings = ServerSettingsService.of({ |
| 227 | start: Effect.void, | 228 | start: Effect.void, |
| 228 | ready: Effect.void, | 229 | ready: Effect.void, |
| 229 | getSettings: Ref.get(settings), | 230 | getSettings: Ref.get(settings).pipe(Effect.tap((value) => Queue.offer(settingsReads, value))), |
| 230 | updateSettings, | 231 | updateSettings, |
| 231 | streamChanges: Stream.fromPubSub(settingsChanges), | 232 | streamChanges: Stream.fromPubSub(settingsChanges), |
| 232 | subscribeChanges: PubSub.subscribe(settingsChanges).pipe( | 233 | subscribeChanges: PubSub.subscribe(settingsChanges).pipe( |
| ... | @@ -271,6 +272,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: | ... | @@ -271,6 +272,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: |
| 271 | snapshots, | 272 | snapshots, |
| 272 | snapshotReadCount, | 273 | snapshotReadCount, |
| 273 | snapshotReads, | 274 | snapshotReads, |
| | 275 | settingsReads, |
| 274 | commands, | 276 | commands, |
| 275 | branchCalls, | 277 | branchCalls, |
| 276 | summaryCalls, | 278 | summaryCalls, |
| ... | @@ -376,6 +378,62 @@ describe("ThreadSettlementReactor", () => { | ... | @@ -376,6 +378,62 @@ describe("ThreadSettlementReactor", () => { |
| 376 | ), | 378 | ), |
| 377 | ); | 379 | ); |
| 378 | | 380 | |
| | 381 | it.effect("skips PR work on startup, timer and merge sweeps when settlement is disabled", () => |
| | 382 | Effect.scoped( |
| | 383 | Effect.gen(function* () { |
| | 384 | yield* TestClock.setTime(Date.parse(NOW)); |
| | 385 | const fixture = yield* makeHarness({ |
| | 386 | snapshot: makeSnapshot([ |
| | 387 | makeThread("branch-thread", { branch: "feature" }), |
| | 388 | makeThread("linked-thread", { |
| | 389 | linkedPullRequest: { |
| | 390 | projectId: PROJECT_ID, |
| | 391 | repository: "owner/repository", |
| | 392 | number: 42, |
| | 393 | url: "https://example.test/owner/repository/pull/42", |
| | 394 | }, |
| | 395 | }), |
| | 396 | ]), |
| | 397 | settings: { |
| | 398 | ...DEFAULT_SERVER_SETTINGS, |
| | 399 | sidebarAutoSettleAfterDays: null, |
| | 400 | sidebarAutoSettleOnMerge: false, |
| | 401 | }, |
| | 402 | }); |
| | 403 | |
| | 404 | yield* Effect.gen(function* () { |
| | 405 | const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; |
| | 406 | yield* reactor.start(); |
| | 407 | yield* Queue.take(fixture.settingsReads); |
| | 408 | yield* Deferred.succeed(fixture.activation, undefined); |
| | 409 | yield* Queue.take(fixture.settingsReads); |
| | 410 | yield* reactor.drain; |
| | 411 | yield* TestClock.adjust("1 minute"); |
| | 412 | yield* Queue.take(fixture.settingsReads); |
| | 413 | yield* reactor.drain; |
| | 414 | yield* fixture.publishMerge; |
| | 415 | yield* Queue.take(fixture.settingsReads); |
| | 416 | yield* reactor.drain; |
| | 417 | |
| | 418 | assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), []); |
| | 419 | assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []); |
| | 420 | assert.deepStrictEqual(yield* Ref.get(fixture.invalidatedCwds), []); |
| | 421 | assert.deepStrictEqual(yield* Ref.get(fixture.commands), []); |
| | 422 | assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 0); |
| | 423 | |
| | 424 | yield* fixture.updateSettings({ sidebarAutoSettleAfterDays: 1 }); |
| | 425 | yield* Queue.take(fixture.snapshotReads); |
| | 426 | yield* reactor.drain; |
| | 427 | // Inactive threads settle without a PR lookup, so only the dispatches prove work resumed. |
| | 428 | assert.deepStrictEqual( |
| | 429 | (yield* Ref.get(fixture.commands)).map((command) => command.threadId).toSorted(), |
| | 430 | [ThreadId.make("branch-thread"), ThreadId.make("linked-thread")], |
| | 431 | ); |
| | 432 | }).pipe(Effect.provide(fixture.layer)); |
| | 433 | }), |
| | 434 | ), |
| | 435 | ); |
| | 436 | |
| 379 | it.effect("starts without clients and skips protected threads before pull request lookup", () => | 437 | it.effect("starts without clients and skips protected threads before pull request lookup", () => |
| 380 | Effect.scoped( | 438 | Effect.scoped( |
| 381 | Effect.gen(function* () { | 439 | Effect.gen(function* () { |
| ... | @@ -694,14 +752,14 @@ describe("ThreadSettlementReactor", () => { | ... | @@ -694,14 +752,14 @@ describe("ThreadSettlementReactor", () => { |
| 694 | Effect.tap((count) => | 752 | Effect.tap((count) => |
| 695 | count === 1 | 753 | count === 1 |
| 696 | ? Deferred.succeed(firstLookupStarted, undefined) | 754 | ? Deferred.succeed(firstLookupStarted, undefined) |
| 697 | : count === 3 | 755 | : count === 2 |
| 698 | ? Deferred.succeed(laterLookupStarted, undefined) | 756 | ? Deferred.succeed(laterLookupStarted, undefined) |
| 699 | : Effect.void, | 757 | : Effect.void, |
| 700 | ), | 758 | ), |
| 701 | Effect.tap((count) => | 759 | Effect.tap((count) => |
| 702 | count === 1 | 760 | count === 1 |
| 703 | ? Deferred.await(releaseFirstLookup) | 761 | ? Deferred.await(releaseFirstLookup) |
| 704 | : count === 3 | 762 | : count === 2 |
| 705 | ? Deferred.await(releaseLaterLookup) | 763 | ? Deferred.await(releaseLaterLookup) |
| 706 | : Effect.void, | 764 | : Effect.void, |
| 707 | ), | 765 | ), |
| ... | @@ -716,13 +774,16 @@ describe("ThreadSettlementReactor", () => { | ... | @@ -716,13 +774,16 @@ describe("ThreadSettlementReactor", () => { |
| 716 | yield* Deferred.succeed(fixture.activation, undefined); | 774 | yield* Deferred.succeed(fixture.activation, undefined); |
| 717 | yield* Queue.take(fixture.snapshotReads); | 775 | yield* Queue.take(fixture.snapshotReads); |
| 718 | yield* Deferred.await(firstLookupStarted); | 776 | yield* Deferred.await(firstLookupStarted); |
| | 777 | yield* Queue.clear(fixture.settingsReads); |
| 719 | | 778 | |
| 720 | yield* fixture.updateSettings({ sidebarAutoSettleOnMerge: false }); | 779 | yield* fixture.updateSettings({ sidebarAutoSettleOnMerge: false }); |
| 721 | yield* Deferred.succeed(releaseFirstLookup, undefined); | 780 | yield* Deferred.succeed(releaseFirstLookup, undefined); |
| 722 | yield* Queue.take(fixture.snapshotReads); | 781 | // The in-flight decision and the newly queued sweep both read the disabled settings. |
| | 782 | yield* Queue.take(fixture.settingsReads); |
| | 783 | yield* Queue.take(fixture.settingsReads); |
| 723 | yield* reactor.drain; | 784 | yield* reactor.drain; |
| 724 | assert.deepStrictEqual(yield* Ref.get(fixture.commands), []); | 785 | assert.deepStrictEqual(yield* Ref.get(fixture.commands), []); |
| 725 | assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 2); | 786 | assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 1); |
| 726 | | 787 | |
| 727 | yield* Ref.set(state, "closed"); | 788 | yield* Ref.set(state, "closed"); |
| 728 | yield* fixture.updateSettings({ enableAgentBrowserAccess: false }); | 789 | yield* fixture.updateSettings({ enableAgentBrowserAccess: false }); |
| ... | @@ -731,8 +792,8 @@ describe("ThreadSettlementReactor", () => { | ... | @@ -731,8 +792,8 @@ describe("ThreadSettlementReactor", () => { |
| 731 | yield* Deferred.succeed(releaseLaterLookup, undefined); | 792 | yield* Deferred.succeed(releaseLaterLookup, undefined); |
| 732 | yield* reactor.drain; | 793 | yield* reactor.drain; |
| 733 | | 794 | |
| 734 | assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 3); | 795 | assert.strictEqual(yield* Ref.get(fixture.snapshotReadCount), 2); |
| 735 | assert.strictEqual(yield* Ref.get(lookupCount), 3); | 796 | assert.strictEqual(yield* Ref.get(lookupCount), 2); |
| 736 | assert.deepStrictEqual( | 797 | assert.deepStrictEqual( |
| 737 | (yield* Ref.get(fixture.commands)).map((command) => command.threadId), | 798 | (yield* Ref.get(fixture.commands)).map((command) => command.threadId), |
| 738 | [ThreadId.make("settings-thread")], | 799 | [ThreadId.make("settings-thread")], |