apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-05-12 08:42:38 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

improve minute alignment logic

Browse source
This commit is contained in:
Ramnique Singh 2025-08-20 16:40:20 +05:30
parent 08e76edd51
commit 0b31585141
3 changed files with 11 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

6
apps/rowboat/src/application/lib/utils/time-to-next-minute.ts Normal file
View file

@ -0,0 +1,6 @@
// returns the number of seconds until the next minute
export function secondsToNextMinute(): number {
const now = new Date();
const secondsUntilNextMinute = 60 - now.getSeconds();
return secondsUntilNextMinute;
}

15
apps/rowboat/src/application/workers/job-rules.worker.ts
View file

@ -8,6 +8,7 @@ import { z } from "zod";
import { nanoid } from "nanoid"; import { nanoid } from "nanoid";
import { PrefixLogger } from "@/app/lib/utils"; import { PrefixLogger } from "@/app/lib/utils";
import { RecurringJobRule } from "@/src/entities/models/recurring-job-rule"; import { RecurringJobRule } from "@/src/entities/models/recurring-job-rule";
import { secondsToNextMinute } from "../lib/utils/time-to-next-minute";
export interface IJobRulesWorker { export interface IJobRulesWorker {
run(): Promise<void>; run(): Promise<void>;
@ -20,8 +21,6 @@ export class JobRulesWorker implements IJobRulesWorker {
private readonly jobsRepository: IJobsRepository; private readonly jobsRepository: IJobsRepository;
private readonly projectsRepository: IProjectsRepository; private readonly projectsRepository: IProjectsRepository;
private readonly pubSubService: IPubSubService; private readonly pubSubService: IPubSubService;
// Run polls aligned to minute marks at this offset (e.g., 2000 ms => :02 each minute)
private readonly minuteAlignmentOffsetMs: number = 2_000;
private workerId: string; private workerId: string;
private logger: PrefixLogger; private logger: PrefixLogger;
private isRunning: boolean = false; private isRunning: boolean = false;
@ -131,14 +130,6 @@ export class JobRulesWorker implements IJobRulesWorker {
} }
} }
// Calculates delay so the next run happens at next minute + minuteAlignmentOffsetMs
private calculateDelayToNextAlignedMinute(): number {
const now = new Date();
const millisecondsUntilNextMinute = (60 - now.getSeconds()) * 1000 - now.getMilliseconds();
const delayMs = millisecondsUntilNextMinute + this.minuteAlignmentOffsetMs;
return delayMs > 0 ? delayMs : this.minuteAlignmentOffsetMs;
}
private async pollScheduled(): Promise<void> { private async pollScheduled(): Promise<void> {
const logger = this.logger.child(`poll-scheduled`); const logger = this.logger.child(`poll-scheduled`);
logger.log("Polling..."); logger.log("Polling...");
@ -176,7 +167,8 @@ export class JobRulesWorker implements IJobRulesWorker {
} }
private scheduleNextPoll(): void { private scheduleNextPoll(): void {
const delayMs = this.calculateDelayToNextAlignedMinute(); // schedule next poll for 2 s past the minute mark
const delayMs = (secondsToNextMinute() + 2) * 1000;
this.logger.log(`Scheduling next poll in ${delayMs} ms`); this.logger.log(`Scheduling next poll in ${delayMs} ms`);
this.pollTimeoutId = setTimeout(async () => { this.pollTimeoutId = setTimeout(async () => {
if (!this.isRunning) return; if (!this.isRunning) return;
@ -195,7 +187,6 @@ export class JobRulesWorker implements IJobRulesWorker {
} }
this.isRunning = true; this.isRunning = true;
this.logger.log(`Starting worker ${this.workerId}`); this.logger.log(`Starting worker ${this.workerId}`);
// No immediate polling; align to 2s past the next minute
this.scheduleNextPoll(); this.scheduleNextPoll();
} }

3
apps/rowboat/src/infrastructure/policies/redis.usage-quota.policy.ts
View file

@ -1,6 +1,7 @@
import { IUsageQuotaPolicy } from "@/src/application/policies/usage-quota.policy.interface"; import { IUsageQuotaPolicy } from "@/src/application/policies/usage-quota.policy.interface";
import { redisClient } from "@/app/lib/redis"; import { redisClient } from "@/app/lib/redis";
import { QuotaExceededError } from "@/src/entities/errors/common"; import { QuotaExceededError } from "@/src/entities/errors/common";
import { secondsToNextMinute } from "@/src/application/lib/utils/time-to-next-minute";
const MAX_QUERIES_PER_MINUTE = Number(process.env.MAX_QUERIES_PER_MINUTE) || 0; const MAX_QUERIES_PER_MINUTE = Number(process.env.MAX_QUERIES_PER_MINUTE) || 0;
@ -15,7 +16,7 @@ export class RedisUsageQuotaPolicy implements IUsageQuotaPolicy {
const count = await redisClient.incr(key); const count = await redisClient.incr(key);
if (count === 1) { if (count === 1) {
await redisClient.expire(key, 70); // Set TTL to clean up automatically await redisClient.expire(key, secondsToNextMinute()); // Set TTL to clean up automatically
} }
if (count > MAX_QUERIES_PER_MINUTE) { if (count > MAX_QUERIES_PER_MINUTE) {