c053f0ba
comments: the concurrency claims are held by alarms now
a73x 2026-08-23 11:13
Commit message
internal/agent/reconcile/worker.go
| Old | New | ||
|---|---|---|---|
| @@ -61,18 +61,9 @@ func (m *manager) deliver(id string, a assignment) { | |||
| 61 | } | 61 | } |
| 62 | 62 | ||
| 63 | // reapAbsent stops the workers for VMs present in neither desired state nor | 63 | // reapAbsent stops the workers for VMs present in neither desired state nor |
| 64 | // local records — nothing is left to reconcile. | 64 | // local records — nothing is left to reconcile. Only IDLE workers are reaped; |
| 65 | // | 65 | // a busy one is deferred to a later tick, which reaping being level-triggered |
| 66 | // Only IDLE workers are reaped. A worker still holding an in-flight (or pending) | 66 | // makes free. |
| 67 | // pass owns that VM, and dropping it from the map would let the next deliver for | ||
| 68 | // the same id spawn a SECOND worker — two goroutines reconciling one VM at once, | ||
| 69 | // which is exactly the invariant this layer exists to provide. A concurrent | ||
| 70 | // Destroy/DeleteVM against a Boot/SaveVM can orphan a hypervisor process with | ||
| 71 | // no record left to find it by, and nothing self-heals from that. Reaping is | ||
| 72 | // level-triggered, so deferring a busy worker to a later tick costs nothing. | ||
| 73 | // This is reachable in ordinary operation: a transiently unreadable record makes | ||
| 74 | // LoadVMs skip a live VM (it continues past read errors), dropping it out of | ||
| 75 | // live for one tick. | ||
| 76 | func (m *manager) reapAbsent(live map[string]assignment) { | 67 | func (m *manager) reapAbsent(live map[string]assignment) { |
| 77 | m.mu.Lock() | 68 | m.mu.Lock() |
| 78 | defer m.mu.Unlock() | 69 | defer m.mu.Unlock() |
| @@ -110,13 +101,6 @@ func (m *manager) tombstones() map[string]bool { | |||
| 110 | // only the worker's lock, which is never held across a pass, so a VM wedged in | 101 | // only the worker's lock, which is never held across a pass, so a VM wedged in |
| 111 | // a multi-second operation cannot delay the report — it simply contributes its | 102 | // a multi-second operation cannot delay the report — it simply contributes its |
| 112 | // previous result, or nothing if it has not published yet. | 103 | // previous result, or nothing if it has not published yet. |
| 113 | // | ||
| 114 | // The rows are CLONED, so every report owns its own protos. A worker that | ||
| 115 | // publishes once and then stays busy is otherwise aliased by every report taken | ||
| 116 | // between two publishes, and proto.Marshal writes a size cache into the message | ||
| 117 | // it marshals — two reports marshalled concurrently would race on one message. | ||
| 118 | // Cloning makes the report self-contained instead of resting on an assumption | ||
| 119 | // about how many goroutines might serialize it. | ||
| 120 | func (m *manager) collect(rep *pb.Report) { | 104 | func (m *manager) collect(rep *pb.Report) { |
| 121 | for _, w := range m.snapshot() { | 105 | for _, w := range m.snapshot() { |
| 122 | w.mu.Lock() | 106 | w.mu.Lock() |
| @@ -217,20 +201,10 @@ func (w *worker) run() { | |||
| 217 | // It takes the same lock the pass-end publish takes and holds it no longer, so | 201 | // It takes the same lock the pass-end publish takes and holds it no longer, so |
| 218 | // a narrating create is no more able to delay a report than a silent one. | 202 | // a narrating create is no more able to delay a report than a silent one. |
| 219 | // | 203 | // |
| 220 | // GENERATION RULE: gen is the generation run stamped on the pass that was | 204 | // gen is the generation run stamped on the pass handed this publisher, and a |
| 221 | // handed this publisher, and a row lands only if that pass is still the one in | 205 | // row lands only if that pass is still the one in flight. Narration reaches |
| 222 | // flight. Narration reaches here on goroutines the pass does not own — the | 206 | // here on goroutines the pass does not own: the image cache shares one download |
| 223 | // image cache shares one download between waiters and runs its progress | 207 | // between waiters and runs its progress callback on the fetch. |
| 224 | // callback on the fetch — so a callback can outlive the pass that asked for | ||
| 225 | // one: a create abandoned at VMTimeout returns and publishes "failed" while the | ||
| 226 | // fetch keeps draining, and one late progress line would otherwise put | ||
| 227 | // "creating — downloading image" back on top of that settled row. The host | ||
| 228 | // report would then show a phantom in-flight create and hide the failure. Both | ||
| 229 | // stale orders die here: a callback after the pass-end publish finds busy | ||
| 230 | // false, and a callback from an earlier pass while a later one runs finds a | ||
| 231 | // newer gen. A callback that beats the pass-end publish is admitted and simply | ||
| 232 | // overwritten by it — the ordering that already made this commentary and not | ||
| 233 | // state: nothing here can outlive the pass that said it. | ||
| 234 | func (w *worker) publish(gen uint64, res vmResult) { | 208 | func (w *worker) publish(gen uint64, res vmResult) { |
| 235 | w.mu.Lock() | 209 | w.mu.Lock() |
| 236 | defer w.mu.Unlock() | 210 | defer w.mu.Unlock() |
internal/server/syncsvc/tracker.go
| Old | New | ||
|---|---|---|---|
| @@ -49,9 +49,6 @@ type netTracker struct { | |||
| 49 | 49 | ||
| 50 | func newNetTracker() *netTracker { return &netTracker{last: map[string]string{}} } | 50 | func newNetTracker() *netTracker { return &netTracker{last: map[string]string{}} } |
| 51 | 51 | ||
| 52 | // writeThrough runs decide→write→commit for one host under a single lock, so | ||
| 53 | // two connections for the same host cannot interleave and leave the cache | ||
| 54 | // disagreeing with the row. | ||
| 55 | func (t *netTracker) writeThrough(hostID, value string, write func() error) error { | 52 | func (t *netTracker) writeThrough(hostID, value string, write func() error) error { |
| 56 | t.mu.Lock() | 53 | t.mu.Lock() |
| 57 | defer t.mu.Unlock() | 54 | defer t.mu.Unlock() |