scheduleBackgroundPushCurrent dedupes and pushes the current local value to the remote store under a lock, without blocking the caller.
()
| 625 | // scheduleBackgroundPushCurrent dedupes and pushes the current local value to the |
| 626 | // remote store under a lock, without blocking the caller. |
| 627 | func (c *counterInt64) scheduleBackgroundPushCurrent() { |
| 628 | // Always mark that a push is needed. |
| 629 | c.bgPushRequested.Store(true) |
| 630 | |
| 631 | // Only one goroutine at a time runs the push loop. |
| 632 | if !c.bgPushInProgress.TryLock() { |
| 633 | return |
| 634 | } |
| 635 | |
| 636 | go func() { |
| 637 | defer func() { |
| 638 | c.bgPushInProgress.Unlock() |
| 639 | // Close the race window: if someone set bgPushRequested between our Swap(false) |
| 640 | // returning false and their TryLock failing, respawn a worker. |
| 641 | if c.bgPushRequested.Load() { |
| 642 | c.scheduleBackgroundPushCurrent() |
| 643 | } |
| 644 | }() |
| 645 | |
| 646 | for { |
| 647 | // Coalesce: if no push is requested at loop start, we're done. |
| 648 | if !c.bgPushRequested.Swap(false) { |
| 649 | return |
| 650 | } |
| 651 | |
| 652 | // Snapshot local state (atomic reads; never block request flow here) |
| 653 | local := c.localState() |
| 654 | // Skip only if UpdatedAt indicates uninitialized; Value can be 0 (e.g., earliest = genesis) |
| 655 | if local.UpdatedAt <= 0 { |
| 656 | continue |
| 657 | } |
| 658 | |
| 659 | // Best-effort fast propagation (no distributed lock): publish the latest state so |
| 660 | // other instances can update their local counters quickly via WatchCounterInt64. |
| 661 | pubCtx, pubCancel := context.WithTimeout(c.registry.appCtx, c.registry.lockMaxWait) |
| 662 | _ = c.registry.connector.PublishCounterInt64(pubCtx, c.key, local) |
| 663 | pubCancel() |
| 664 | |
| 665 | // Acquire distributed lock with a bounded wait budget. |
| 666 | // NOTE: This is a background operation and MUST NOT block request flow. |
| 667 | lockCtx, lockCancel := context.WithTimeout(c.registry.appCtx, c.registry.lockMaxWait) |
| 668 | unlock := c.tryAcquireLock(lockCtx) |
| 669 | lockCancel() |
| 670 | if unlock == nil { |
| 671 | // Could not acquire lock quickly; rely on publish-only propagation for now. |
| 672 | // If newer updates arrive, scheduleBackgroundPushCurrent() will set bgPushRequested |
| 673 | // and this loop will run again. |
| 674 | continue |
| 675 | } |
| 676 | |
| 677 | func() { |
| 678 | defer unlock() |
| 679 | |
| 680 | // Reconcile with remote under the distributed lock. |
| 681 | getCtx, getCancel := context.WithTimeout(c.registry.appCtx, c.registry.fallbackTimeout) |
| 682 | remote, remoteOk := c.tryGetRemoteState(getCtx) |
| 683 | getCancel() |
| 684 |