Fix/db storage - #111
Fix/db storage#111Amazing-Stardom wants to merge 22 commits into
Conversation
Offload Code Diffs to Blob StorageOverviewThis change introduces an automated archival system to move raw code diffs from PostgreSQL metadata to external blob storage after thirty days. It adds background workers, API endpoints, and a migration tool to prevent database bloat and reduce backup sizes. Technical Highlights
Impact
|
Offload Code Diffs to Blob StorageOverviewThis architectural update moves old code diffs from PostgreSQL JSONB metadata to external blob storage. It adds background workers, API endpoints, and configuration settings to automate the archival process. Technical Highlights
Impact
|
| } | ||
| defer r.Close() | ||
|
|
||
| size := r.Size() |
There was a problem hiding this comment.
Severity: warning
The r.Size() method may not provide accurate results or might not be available for all blob storage backends or implementations.
Suggestions:
- Add a comment explaining the reliance on
r.Size()and the fallbackio.LimitReaderfor robustness. - Consider if
io.LimitReaderalone is sufficient, or ifr.Size()is critical for early exit.
| effectiveCronExpr := "30 21 * * *" | ||
| var archivalRetentionDays int = 30 | ||
|
|
||
| if db != nil { |
There was a problem hiding this comment.
Severity: warning
The database query is using context.Background(). Please ensure that this is acceptable for long-running startup operations, as it might not propagate cancellation signals.
Suggestions:
- Consider passing a context with a timeout if the DB query could hang indefinitely during startup.
|
|
||
| if db != nil { | ||
| var data []byte | ||
| err := db.QueryRowContext(context.Background(), "SELECT data FROM system_settings WHERE name = 'preloaded_changes_archival_settings'").Scan(&data) |
There was a problem hiding this comment.
Severity: warning
Potential sql.ErrNoRows is not explicitly handled. The check err == nil might be misleading if no row actually exists in the database.
Suggestions:
- Explicitly check for
sql.ErrNoRowsand handle it to clarify intent, though current logic defaults correctly.
| RetentionDays int `json:"retention_days"` | ||
| } | ||
| if unmarshalErr := json.Unmarshal(data, &cfg); unmarshalErr != nil { | ||
| log.Printf("[jobqueue] failed to unmarshal archival settings from DB: %v", unmarshalErr) |
There was a problem hiding this comment.
Severity: warning
The log message for the unmarshal error lacks context. It would be beneficial to include the name of the setting that failed to unmarshal.
Suggestions:
- Add
name(e.g.,preloaded_changes_archival_settings) to the log message for better debuggability.
|
|
||
| parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor) | ||
| archivalSchedule, parseErr := parser.Parse(effectiveCronExpr) | ||
| if parseErr != nil { |
There was a problem hiding this comment.
Severity: info
The log message for the cron parse error uses Printf instead of Errorf. This might cause the error to be missed or not properly highlighted in the logs.
Suggestions:
- Use
log.Errorfor a structured logger with error level for cron parse failures.
| log.Printf("[jobqueue] preloaded_changes archival cron=%q next_run=%s retention_days=%d", effectiveCronExpr, nextRun.Format("2006-01-02 15:04:05 MST"), archivalRetentionDays) | ||
| } | ||
|
|
||
| // Verification 1: Ensure River database schema matches the River Go library version |
There was a problem hiding this comment.
Severity: info
This is an excellent addition. River schema validation effectively prevents silent data corruption that could arise from missed database migrations.
| CompletedJobRetentionPeriod: 365 * 24 * time.Hour, | ||
| CancelledJobRetentionPeriod: 365 * 24 * time.Hour, | ||
| DiscardedJobRetentionPeriod: 365 * 24 * time.Hour, | ||
| CompletedJobRetentionPeriod: 30 * 24 * time.Hour, |
There was a problem hiding this comment.
Severity: info
Reducing the job retention period from 365 days to 30 days will significantly decrease the database size and improve overall performance.
| func() (river.JobArgs, *river.InsertOpts) { | ||
| return PreloadedChangesArchivalSweepJobArgs{ | ||
| RetentionDays: archivalRetentionDays, | ||
| BatchSize: 0, |
There was a problem hiding this comment.
Severity: warning
Setting BatchSize: 0 is unclear regarding its intent. It could imply the default batch size or no batching at all. Please clarify the intended behavior or use a meaningful default value.
Suggestions:
- Add a comment explaining
BatchSize: 0behavior, or set a sensible default if it's not handled by the worker.
| }, | ||
| &river.PeriodicJobOpts{ | ||
| ID: "preloaded_changes_archival_sweep", | ||
| RunOnStart: true, |
There was a problem hiding this comment.
Severity: info
Setting RunOnStart: true for this periodic job ensures that it will execute immediately after the application starts up.
| InsertOpts: &river.InsertOpts{ | ||
| Queue: "preloaded_changes_archival", | ||
| UniqueOpts: river.UniqueOpts{ | ||
| ByArgs: true, |
There was a problem hiding this comment.
Severity: info
Using UniqueOpts{ByArgs: true} is a good practice for preventing the creation of duplicate archival jobs.
| } | ||
| res, err := jq.client.InsertMany(ctx, params) | ||
| if err != nil { | ||
| log.Printf("[ERROR] Failed to queue preloaded_changes archival jobs: %v", err) |
There was a problem hiding this comment.
Severity: info
The log message uses Printf instead of Errorf. This might cause the log entry to be overlooked or not treated as an error.
Suggestions:
- Use
log.Errorfor a structured logger with error level for failed job queuing.
| // UpdateArchivalSchedule dynamically updates the River periodic schedule for preloaded_changes archival. | ||
| func (jq *JobQueue) UpdateArchivalSchedule(cronExpr string) error { | ||
| if jq == nil || jq.client == nil { | ||
| return nil |
There was a problem hiding this comment.
Severity: warning
Returning nil when jq or client is nil might silently ignore schedule updates. It's important to ensure these cases are handled appropriately.
Suggestions:
- Return an error here, similar to
EnqueuePreloadedChangesArchivalSweep, to indicate a misconfigured job queue.
| return fmt.Errorf("invalid cron expression %q: %w", cronExpr, err) | ||
| } | ||
| // Remove any existing schedule for this job before adding the new one dynamically | ||
| jq.client.PeriodicJobs().RemoveByID("preloaded_changes_archival_sweep") |
There was a problem hiding this comment.
Severity: info
Calling RemoveByID before Add is a crucial step for correctly updating dynamic schedules.
| schedule, | ||
| func() (river.JobArgs, *river.InsertOpts) { | ||
| return PreloadedChangesArchivalSweepJobArgs{ | ||
| RetentionDays: 30, // Using default here; the sweep worker fetches the live value dynamically |
There was a problem hiding this comment.
Severity: info
The RetentionDays is hardcoded to 30 here, but the comment suggests that the worker fetches the live value. Please confirm that the worker logic correctly handles this and that there is no discrepancy.
Suggestions:
- Ensure
preloadedChangesArchivalSweepWorkerindeed fetches the current retention days from settings, making this hardcoded value harmless.
| }, | ||
| &river.PeriodicJobOpts{ | ||
| ID: "preloaded_changes_archival_sweep", | ||
| RunOnStart: true, |
There was a problem hiding this comment.
Severity: info
Setting RunOnStart: true for this dynamically added periodic job ensures that it takes effect immediately upon startup.
| AND metadata ? 'preloaded_changes' | ||
| ORDER BY created_at ASC; | ||
| ` | ||
| rows, err := w.db.QueryContext(ctx, query, retentionDays) |
There was a problem hiding this comment.
Severity: warning
The sweep query fetches all eligible reviews at once. This could lead to a memory issue if there are millions of records to process.
Suggestions:
- Implement batching for
archivalJobscreation, e.g., fetch and enqueue in chunks. - Consider using
COPY (SELECT ...) TO STDOUTfor very large datasets ifw.db.QueryContextbecomes a bottleneck.
| log.Error().Err(uploadErr).Int64("review_id", reviewID).Msg("[preloaded_changes_archival_worker] blob upload failed") | ||
|
|
||
| // If it's a fatal write permission error, cancel the job permanently so it doesn't infinitely retry | ||
| if strings.Contains(strings.ToLower(uploadErr.Error()), "permission denied") { |
There was a problem hiding this comment.
Severity: info
This is good. Calling JobCancel for fatal permission errors effectively prevents infinite retry loops.
| return fmt.Errorf("failed to query pending jobs for batch %s: %w", batchRunID, err) | ||
| } | ||
| if pendingCount > 0 { | ||
| log.Warn().Str("batch_run_id", batchRunID).Int("pending", pendingCount).Msg("[preloaded_changes_purge_worker] 6 hours passed with jobs still pending — discarding purge job (no data deleted)") |
There was a problem hiding this comment.
Severity: critical
This is a critical safety measure. Calling JobCancel prevents data purging if jobs remain pending after a 6-hour timeout.
| log.Info().Str("batch_run_id", batchRunID).Int("pending", pendingCount).Str("snooze", snoozeDuration.String()).Str("elapsed", elapsed.Round(time.Second).String()).Msg("[preloaded_changes_purge_worker] jobs still pending, snoozing with exponential backoff...") | ||
| // JobSnooze safely reschedules the job without counting it as a failure, | ||
| // and automatically extends MaxAttempts so it never permanently dies from snoozing. | ||
| return river.JobSnooze(snoozeDuration) |
There was a problem hiding this comment.
Severity: info
This is good. Using JobSnooze implements intelligent backoff and avoids unnecessarily marking jobs as failures.
| } | ||
|
|
||
| // 2. O(1) Fetch & Update (Postgres does all the work internally) | ||
| purgeQuery := ` |
There was a problem hiding this comment.
Severity: info
This is good. The bulk UPDATE statement uses a subquery for efficiency, and the metadata - 'key' syntax is used for JSONB removal.
| type PreloadedChangesArchivalConfig struct { | ||
| // MaxWorkers is the concurrency of the "preloaded_changes_archival" queue. | ||
| MaxWorkers int // default: 10 | ||
| // BatchSize is the number of reviews packaged into a single archival job. |
There was a problem hiding this comment.
Severity: warning
The description for BatchSize is unclear. While individual archival jobs process one review, the sweep job does not utilize this BatchSize for fetching records. Please clarify the intended use or remove the ambiguity.
Suggestions:
- Clarify
BatchSizepurpose: if it's for the sweep, implement its usage there. If it's for individual archival jobs, clarify how it applies (e.g., if a single archival job could process multiple reviews, which is not currently the case). - Consider renaming
BatchSizeif it's not used for batching reviews, or remove if unused.
| MaxWorkers: 10, | ||
| BatchSize: 10, | ||
| } | ||
| if v := os.Getenv("LIVEREVIEW_PRELOADED_CHANGES_ARCHIVAL_MAX_WORKERS"); v != "" { |
There was a problem hiding this comment.
Severity: info
This is good. Environment variables can be used to override MaxWorkers and BatchSize, providing flexibility in configuration.
| if n, err := strconv.Atoi(v); err == nil && n > 0 { | ||
| config.MaxWorkers = n | ||
| } | ||
| } else if v := os.Getenv("LIVEREVIEW_DIFF_ARCHIVAL_MAX_WORKERS"); v != "" { |
There was a problem hiding this comment.
Severity: info
This is good. The implementation includes backward compatibility for older environment variable names, ensuring a smoother transition.
| modelDiffsPayload, err := json.Marshal(modelDiffs) | ||
| if err != nil { | ||
| w.handleFailure(ctx, args, logger, eventSink, fmt.Sprintf("failed to marshal diffs: %v", err), "failed_to_marshal_diffs") | ||
| return nil |
There was a problem hiding this comment.
Severity: warning
An early return in this function skips the logger cleanup and leaves the review in a running state. Please use handleFailure properly or ensure that cleanup is performed.
Suggestions:
- Ensure eventSink and logger are properly closed or failed before returning nil on marshal error
| <div className="grid grid-cols-6 gap-1 w-fit"> | ||
| {HOURS_12.map((h12) => { | ||
| const h24 = h12 === 0 ? offset : h12 + offset; // 12AM=0, 1AM=1...11AM=11, 12PM=12, 1PM=13...11PM=23 | ||
| const isSelected = hours.includes(h24 === 24 ? 12 : h24); // handle edge: 12+12=24 shouldn't happen since offset is 0 or 12 |
There was a problem hiding this comment.
Severity: warning
With offsets of 0 or 12, it's impossible to achieve h24=24. This appears to be a dead check, as h24 === 24 will never be true under these conditions.
Suggestions:
- Remove the redundant
h24 === 24check
| return enrichedFiles; | ||
| }, [enrichedFiles, sortMode, canSortByRisk]); | ||
|
|
||
| const facets = useMemo(() => buildFilterFacets(files, filters), [files, filters]); |
There was a problem hiding this comment.
Severity: warning
The buildFilterFacets function has been updated to use enrichedFiles instead of the previous files.
Suggestions:
- Confirm
enrichedFilesalways contains necessary data for facet building. IffilesandenrichedFilesare not always identical in structure/content, this could lead to subtle filtering bugs or missing facets.
| <table className="w-full table-fixed border-collapse font-mono text-xs"> | ||
| <tbody> | ||
| {lines.map((line, idx) => { | ||
| const lineComments = comments.filter(({ comment }) => commentBelongsToLine(comment, line) && commentMatchesFilters(comment, filters)); |
There was a problem hiding this comment.
Severity: info
The commentBelongsToLine function now accepts lines as an argument.
Suggestions:
- Ensure
commentBelongsToLinecorrectly handles all edge cases with the addedlinescontext, especially for comments spanning multiple lines or at hunk boundaries.
|
|
||
| useEffect(() => clearTimers, [clearTimers]); | ||
|
|
||
| const isActive = vote === 'up' || vote === 'down'; |
There was a problem hiding this comment.
Severity: info
The postFeedback useCallback has been removed.
Suggestions:
- Ensure feedback submission logic is correctly reimplemented elsewhere or confirm it's no longer needed. Missing this could lead to silent failures in feedback collection.
| } | ||
| }, [feedbackId]); | ||
|
|
||
| const calculatePos = (wrapperEl: HTMLElement, popupEl?: HTMLElement | null) => { |
There was a problem hiding this comment.
Severity: info
A new calculatePos function has been introduced for handling popup positioning.
Suggestions:
- Consider adding comments to
calculatePosto explain the positioning logic, especially viewport edge handling.
| setPopupVisible(false); | ||
| setPopupMode(null); | ||
| setStatsExpanded(false); | ||
| }, 280); |
There was a problem hiding this comment.
Severity: warning
The setStatsExpanded(false) call has been removed. Please ensure that this state is managed elsewhere if it is still required.
Suggestions:
- Confirm
setStatsExpandedis no longer required or is handled by a different mechanism.
4 participants
Fix: DB Storage Growth from River Job Retention and preloaded_changes Metadata
TL;DR
PostgreSQL storage grew by 44% due to two root causes.
First, the River job queue had no retention limit, so completed and cancelled jobs accumulated in the
river_jobtable indefinitely.Second, large JSON blobs in the
preloaded_changesfield of thereviews.metadatacolumn were never removed after reviews were completed.Changes
River job retention limits
Set River job retention periods to prevent unbounded table growth:
Preloaded changes archival cron job
Added a background cron job that runs daily at 21:30 UTC (3:00 AM IST) to move
preloaded_changesJSON from PostgreSQL to blob storage.The job works as follows:
preloaded_changesin metadata.preloaded_changesfrom the metadata column in bulk.An admin settings tab at Settings/Storage allows operators to configure the schedule and retention period and trigger a manual run.
Fix: old review results failing to load
Very old CLI reviews stored the
confidencefield as a number (e.g.0.95). The backend struct expected a string. Loading those reviews returned this error:The fix runs a second decode pass only when this specific error occurs. It converts any numeric
confidencevalue to a string and retries. New reviews are not affected.Validation
5645(an old CLI review with numeric confidence) and confirmed it loads without error.go build ./internal/api ./internal/jobqueuepasses with 0 errors.npx tsc --noEmitpasses with 0 errors.Checklist
SECURITY.mdand updated related documentation if needed.Notes For Reviewers
NextRetrypattern. It reschedules itself every 5 minutes until all archival jobs for a batch complete. Review this flow inPreloadedChangesArchivalPurgeWorker.decodeReviewResultis intentionally narrow. It only runs when the error message contains"cannot unmarshal number".