nathan/vctp2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

work on daily aggregation with postgresql
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Nathan Coad
2026-02-11 10:48:32 +11:00
parent a0556e3ac8
commit b5bcea9da5
1 changed files with 25 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
internal/tasks/dailyAggregate.go
View File

@@ -3,6 +3,7 @@ package tasks
import (
"context"
"database/sql"
"errors"
"fmt"
"log/slog"
"os"
@@ -406,14 +407,33 @@ LIMIT 1
}
// Refine lifecycle with existing SQL helper to pick up first-after deletions.
if err := db.RefineCreationDeletionFromUnion(ctx, dbConn, summaryTable, unionQuery); err != nil {
c.Logger.Warn("failed to refine creation/deletion times", "error", err, "table", summaryTable)
refineStart := time.Now()
c.Logger.Debug("Refining creation/deletion times", "table", summaryTable)
refineCtx, cancelRefine := context.WithTimeout(ctx, 2*time.Minute)
defer cancelRefine()
if err := db.RefineCreationDeletionFromUnion(refineCtx, dbConn, summaryTable, unionQuery); err != nil {
if errors.Is(err, context.DeadlineExceeded) || errors.Is(refineCtx.Err(), context.DeadlineExceeded) {
c.Logger.Warn("timed out refining creation/deletion times; continuing", "table", summaryTable, "timeout", "2m")
} else {
c.Logger.Debug("refined creation/deletion times", "table", summaryTable)
c.Logger.Warn("failed to refine creation/deletion times", "error", err, "table", summaryTable)
}
if err := db.UpdateSummaryPresenceByWindow(ctx, dbConn, summaryTable, dayStart.Unix(), dayEnd.Unix()); err != nil {
} else {
c.Logger.Debug("Refined creation/deletion times", "table", summaryTable, "duration", time.Since(refineStart))
}
presenceStart := time.Now()
c.Logger.Debug("Updating daily AvgIsPresent from lifecycle window", "table", summaryTable)
presenceCtx, cancelPresence := context.WithTimeout(ctx, 2*time.Minute)
defer cancelPresence()
if err := db.UpdateSummaryPresenceByWindow(presenceCtx, dbConn, summaryTable, dayStart.Unix(), dayEnd.Unix()); err != nil {
if errors.Is(err, context.DeadlineExceeded) || errors.Is(presenceCtx.Err(), context.DeadlineExceeded) {
c.Logger.Warn("timed out updating daily AvgIsPresent from lifecycle window; continuing", "table", summaryTable, "timeout", "2m")
} else {
c.Logger.Warn("failed to update daily AvgIsPresent from lifecycle window (Go path)", "error", err, "table", summaryTable)
}
} else {
c.Logger.Debug("Updated daily AvgIsPresent from lifecycle window", "table", summaryTable, "duration", time.Since(presenceStart))
}
analyzeStart := time.Now()
c.Logger.Debug("Analyzing daily summary table", "table", summaryTable)