RHINENG-26348: add job to refill 0 partition - #2289
Open
TenSt wants to merge 2 commits into
Open
Conversation
Reviewer's GuideImplements a one-shot recovery job that publishes recalc events for non-stale bucket-0 systems with advisory notifications skipped, adds skip-notification semantics end-to-end in evaluator and platform events, and wires the job into clowdapp configuration and tests. File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- In writeEvents/writeEventsSkipNotifications, reqs[acc][start:end] assumes that request IDs exist for every account; for recovery EvalDataSlice entries where RequestID is empty this will leave reqs[acc] nil and can panic when slicing, so it would be safer to guard this access (e.g., use nil or an empty slice when reqs[acc] is nil or shorter than invs).
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In writeEvents/writeEventsSkipNotifications, reqs[acc][start:end] assumes that request IDs exist for every account; for recovery EvalDataSlice entries where RequestID is empty this will leave reqs[acc] nil and can panic when slicing, so it would be safer to guard this access (e.g., use nil or an empty slice when reqs[acc] is nil or shorter than invs).
## Individual Comments
### Comment 1
<location path="tasks/system_advisories_0_recovery/recovery.go" line_range="63-72" />
<code_context>
+ return nil
+}
+
+func getNonStaleBucket0InventoryIDs() ([]mqueue.EvalData, error) {
+ var inventoryAIDs []mqueue.EvalData
+ err := tasks.CancelableDB().Table("system_inventory si").
+ Select("si.inventory_id, si.rh_account_id, ra.org_id").
+ Joins("JOIN rh_account ra ON ra.id = si.rh_account_id").
+ Where("si.stale = false").
+ Where("satisfies_hash_partition('system_advisories'::regclass, ?, ?, si.rh_account_id)",
+ systemAdvisoriesPartitions, systemAdvisoriesRemainder).
+ Order("si.rh_account_id, si.id").
+ Scan(&inventoryAIDs).Error
+ if err != nil {
+ return nil, err
+ }
+ return inventoryAIDs, nil
+}
</code_context>
<issue_to_address>
**suggestion (performance):** Loading all bucket-0 inventory IDs into memory may not scale well for large accounts.
This function reads the entire non-stale bucket-0 result set into memory before processing. For large accounts this can create high memory usage and delay publishing until the full scan finishes. Since this is a recovery job that may run on production-scale data, consider streaming in batches (e.g., paginating by primary key and calling `WriteEventsSkipNotifications` per batch) to cap memory usage and begin sending events earlier.
Suggested implementation:
```golang
var (
totalCount int
lastInventoryID int64
)
for {
inventoryAIDs, err := getNonStaleBucket0InventoryIDs(lastInventoryID, recoveryBatchSize)
if err != nil {
utils.LogError("err", err, "failed to load non-stale bucket-0 inventory IDs")
return err
}
if len(inventoryAIDs) == 0 {
break
}
err = mqueue.EvalDataSlice(inventoryAIDs).WriteEventsSkipNotifications(base.Context, evalWriter, recoveryBatchSize)
if err != nil {
utils.LogError("err", err, "sending recovery recalc messages failed")
return err
}
totalCount += len(inventoryAIDs)
// advance pagination cursor by the last inventory ID in this batch
lastInventoryID = inventoryAIDs[len(inventoryAIDs)-1].InventoryID
}
utils.LogInfo("count", totalCount, "seconds", time.Since(start).Seconds(),
"systems sent to recovery recalc with skip_notifications")
return nil
}
```
```golang
func getNonStaleBucket0InventoryIDs(afterInventoryID int64, limit int) ([]mqueue.EvalData, error) {
var inventoryAIDs []mqueue.EvalData
db := tasks.CancelableDB().Table("system_inventory si").
Select("si.inventory_id, si.rh_account_id, ra.org_id").
Joins("JOIN rh_account ra ON ra.id = si.rh_account_id").
Where("si.stale = false").
Where("satisfies_hash_partition('system_advisories'::regclass, ?, ?, si.rh_account_id)",
systemAdvisoriesPartitions, systemAdvisoriesRemainder).
// paginate by primary key / stable ordering to allow streaming in batches
Order("si.inventory_id").
Limit(limit)
if afterInventoryID > 0 {
db = db.Where("si.inventory_id > ?", afterInventoryID)
}
if err := db.Scan(&inventoryAIDs).Error; err != nil {
return nil, err
}
return inventoryAIDs, nil
}
```
1. This change assumes that `mqueue.EvalData` has an `InventoryID` field matching `si.inventory_id`; if the field is named differently, adjust `inventoryAIDs[len(inventoryAIDs)-1].InventoryID` accordingly.
2. If `system_inventory`'s stable pagination column is not `inventory_id`, replace both the `Order("si.inventory_id")` and `Where("si.inventory_id > ?", afterInventoryID)` with the correct primary key column.
3. If there are other call sites for `getNonStaleBucket0InventoryIDs`, update them to pass the `afterInventoryID` cursor and `limit` (e.g., usually `0` and `recoveryBatchSize`) or introduce a thin wrapper for backwards compatibility.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #2289 +/- ##
==========================================
- Coverage 59.27% 59.12% -0.15%
==========================================
Files 148 149 +1
Lines 9488 9549 +61
==========================================
+ Hits 5624 5646 +22
- Misses 3276 3311 +35
- Partials 588 592 +4
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
TenSt
force-pushed
the
stepan/RHINENG-26348-add-job-to-refill-0-partition
branch
from
July 30, 2026 17:26
fc6e383 to
52897ba
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR:
system_advisories_0_recoveryjob to recalc non-stale bucket-0 systemspatchman.evaluator.recalcin 500/msg chunks withskip_notificationsJOBS_CONFIG+ disabled CJI (SA0_RECOVERY_DISABLED, default true)Note: I've tested locally and all looks good. We will tests everything in stage before going to production even that stage doesn't have the corrupted partition.
Summary by Sourcery
Introduce a one-shot recovery job to recalc non-stale bucket-0 systems while marking advisories as notified without sending instant notifications, wired through evaluator events and deployment config.
New Features:
Enhancements:
Build:
Tests: