| ... | @@ -7,6 +7,7 @@ import * as FileSystem from "effect/FileSystem"; | ... | @@ -7,6 +7,7 @@ import * as FileSystem from "effect/FileSystem"; |
| 7 | import * as Layer from "effect/Layer"; | 7 | import * as Layer from "effect/Layer"; |
| 8 | import * as PubSub from "effect/PubSub"; | 8 | import * as PubSub from "effect/PubSub"; |
| 9 | import * as Ref from "effect/Ref"; | 9 | import * as Ref from "effect/Ref"; |
| | 10 | import * as Schedule from "effect/Schedule"; |
| 10 | import * as Scope from "effect/Scope"; | 11 | import * as Scope from "effect/Scope"; |
| 11 | import * as Stream from "effect/Stream"; | 12 | import * as Stream from "effect/Stream"; |
| 12 | import * as SynchronizedRef from "effect/SynchronizedRef"; | 13 | import * as SynchronizedRef from "effect/SynchronizedRef"; |
| ... | @@ -23,6 +24,8 @@ import { mergeGitStatusParts } from "@t3tools/shared/git"; | ... | @@ -23,6 +24,8 @@ import { mergeGitStatusParts } from "@t3tools/shared/git"; |
| 23 | import * as GitWorkflowService from "../git/GitWorkflowService.ts"; | 24 | import * as GitWorkflowService from "../git/GitWorkflowService.ts"; |
| 24 | | 25 | |
| 25 | const DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30); | 26 | const DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30); |
| | 27 | const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30); |
| | 28 | const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15); |
| 26 | | 29 | |
| 27 | interface VcsStatusChange { | 30 | interface VcsStatusChange { |
| 28 | readonly cwd: string; | 31 | readonly cwd: string; |
| ... | @@ -48,6 +51,20 @@ interface StreamStatusOptions { | ... | @@ -48,6 +51,20 @@ interface StreamStatusOptions { |
| 48 | readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>; | 51 | readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>; |
| 49 | } | 52 | } |
| 50 | | 53 | |
| | 54 | export function remoteRefreshFailureDelay( |
| | 55 | consecutiveFailures: number, |
| | 56 | configuredInterval: Duration.Duration, |
| | 57 | ) { |
| | 58 | const exponent = Math.max(0, consecutiveFailures - 1); |
| | 59 | const backoffMs = |
| | 60 | Duration.toMillis(VCS_STATUS_REFRESH_FAILURE_BASE_DELAY) * Math.pow(2, exponent); |
| | 61 | const cappedBackoff = Duration.min( |
| | 62 | Duration.millis(backoffMs), |
| | 63 | VCS_STATUS_REFRESH_FAILURE_MAX_DELAY, |
| | 64 | ); |
| | 65 | return Duration.max(configuredInterval, cappedBackoff); |
| | 66 | } |
| | 67 | |
| 51 | export interface VcsStatusBroadcasterShape { | 68 | export interface VcsStatusBroadcasterShape { |
| 52 | readonly getStatus: ( | 69 | readonly getStatus: ( |
| 53 | input: VcsStatusInput, | 70 | input: VcsStatusInput, |
| ... | @@ -241,32 +258,46 @@ export const layer = Layer.effect( | ... | @@ -241,32 +258,46 @@ export const layer = Layer.effect( |
| 241 | cwd: string, | 258 | cwd: string, |
| 242 | automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>, | 259 | automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>, |
| 243 | ) => { | 260 | ) => { |
| 244 | const logRefreshFailure = (error: GitManagerServiceError) => | 261 | return Effect.gen(function* () { |
| 245 | Effect.logWarning("VCS remote status refresh failed", { | 262 | const consecutiveFailuresRef = yield* Ref.make(0); |
| 246 | cwd, | 263 | const refreshRemoteStatusIfEnabled = Effect.gen(function* () { |
| 247 | detail: error.message, | 264 | const configuredInterval = yield* automaticRemoteRefreshInterval; |
| | 265 | const activeInterval = Duration.isZero(configuredInterval) |
| | 266 | ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL |
| | 267 | : configuredInterval; |
| | 268 | if (Duration.isZero(configuredInterval)) { |
| | 269 | return activeInterval; |
| | 270 | } |
| | 271 | |
| | 272 | const exit = yield* refreshRemoteStatus(cwd).pipe(Effect.exit); |
| | 273 | if (Exit.isSuccess(exit)) { |
| | 274 | yield* Ref.set(consecutiveFailuresRef, 0); |
| | 275 | return activeInterval; |
| | 276 | } |
| | 277 | |
| | 278 | const consecutiveFailures = yield* Ref.updateAndGet( |
| | 279 | consecutiveFailuresRef, |
| | 280 | (count) => count + 1, |
| | 281 | ); |
| | 282 | const nextDelay = remoteRefreshFailureDelay(consecutiveFailures, activeInterval); |
| | 283 | yield* Effect.logWarning("VCS remote status refresh failed", { |
| | 284 | cwd, |
| | 285 | detail: exit.cause.toString(), |
| | 286 | consecutiveFailures, |
| | 287 | nextDelayMs: Duration.toMillis(nextDelay), |
| | 288 | }); |
| | 289 | return nextDelay; |
| 248 | }); | 290 | }); |
| 249 | const refreshRemoteStatusIfEnabled = automaticRemoteRefreshInterval.pipe( | | |
| 250 | Effect.flatMap((interval) => | | |
| 251 | Duration.isZero(interval) ? Effect.void : refreshRemoteStatus(cwd).pipe(Effect.asVoid), | | |
| 252 | ), | | |
| 253 | ); | | |
| 254 | const sleepForConfiguredInterval = automaticRemoteRefreshInterval.pipe( | | |
| 255 | Effect.flatMap((interval) => | | |
| 256 | Effect.sleep(Duration.isZero(interval) ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL : interval), | | |
| 257 | ), | | |
| 258 | ); | | |
| 259 | | 291 | |
| 260 | return refreshRemoteStatusIfEnabled.pipe( | 292 | return yield* refreshRemoteStatusIfEnabled.pipe( |
| 261 | Effect.catch(logRefreshFailure), | 293 | Effect.repeat( |
| 262 | Effect.andThen( | 294 | Schedule.identity<Duration.Duration>().pipe( |
| 263 | Effect.forever( | 295 | Schedule.addDelay((delay) => Effect.succeed(delay)), |
| 264 | sleepForConfiguredInterval.pipe( | | |
| 265 | Effect.andThen(refreshRemoteStatusIfEnabled.pipe(Effect.catch(logRefreshFailure))), | | |
| 266 | ), | 296 | ), |
| 267 | ), | 297 | ), |
| 268 | ), | 298 | Effect.asVoid, |
| 269 | ); | 299 | ); |
| | 300 | }); |
| 270 | }; | 301 | }; |
| 271 | | 302 | |
| 272 | const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* ( | 303 | const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* ( |