Preserve absolute event timestamps in Gossip by phanky1 · Pull Request #10418 · celery/celery

phanky1 · 2026-07-17T11:55:30Z

Description

Fixes: #6213.

Worker event timestamps are Unix epochs and are already independent of timezone: the heartbeat stamps events with time.time()

celery/celery/events/event.py

Lines 18 to 27 in d96df92

    
           def Event(type, _fields=None, __dict__=dict, __now__=time.time, **fields): 
        
               """Create an event. 
        
               Notes: 
        
                   An event is simply a dictionary: the only required field is ``type``. 
        
                   A ``timestamp`` field will be set to the current time if not provided. 
        
               """ 
        
               event = __dict__(_fields, **fields) if _fields else fields 
        
               if 'timestamp' not in event: 
        
                   event.update(timestamp=__now__(), type=type)

and Gossip records the receive time the same way through event_from_message's now=time.time default

celery/celery/events/receiver.py

Lines 113 to 137 in d96df92

    
           def event_from_message(self, body, localize=True, 
        
                                  now=time.time, tzfields=_TZGETTER, 
        
                                  adjust_timestamp=adjust_timestamp, 
        
                                  CLIENT_CLOCK_SKEW=CLIENT_CLOCK_SKEW): 
        
               type = body['type'] 
        
               if type == 'task-sent': 
        
                   # clients never sync so cannot use their clock value 
        
                   _c = body['clock'] = (self.clock.value or 1) + CLIENT_CLOCK_SKEW 
        
                   self.adjust_clock(_c) 
        
               else: 
        
                   try: 
        
                       clock = body['clock'] 
        
                   except KeyError: 
        
                       body['clock'] = self.forward_clock() 
        
                   else: 
        
                       self.adjust_clock(clock) 
        
               if localize: 
        
                   try: 
        
                       offset, timestamp = tzfields(body) 
        
                   except KeyError: 
        
                       pass 
        
                   else: 
        
                       body['timestamp'] = adjust_timestamp(timestamp, offset) 
        
               body['local_received'] = now()

The two values are directly comparable epochs.

However, Gossip currently uses event_from_message with its default localize=True, which shifts received timestamps by the difference between the sender's and receiver's timezone offsets, thus generating this false warning:

[2020-07-02 14:01:58,138: WARNING/MainProcess] Substantial drift from celery@worker-two may mean clocks are out of sync.  Current drift is 25200 seconds.  [orig: 2020-07-02 14:01:58.138430 recv: 2020-07-02 21:01:58.134958]

This doesn't happen in Django because Django sets the same TZ environment variable for all processes.

codecov · 2026-07-17T12:00:15Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 88.43%. Comparing base (f85031f) to head (bb28259).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #10418   +/-   ##
=======================================
  Coverage   88.43%   88.43%           
=======================================
  Files         153      153           
  Lines       19802    19802           
  Branches     2301     2301           
=======================================
+ Hits        17511    17512    +1     
  Misses       1994     1994           
+ Partials      297      296    -1

Flag	Coverage Δ
unittests	`88.41% <ø> (+<0.01%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

Copilot

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

…bile fixes Examples: replace all four language examples with structurally impressive, multi-finding live PRs found via api.striff.io on the fixed C# parser stack: apache/pinot#19073 (Java, first-ever core-to-plugin edge, Copilot contrast), celery/celery#10418 (Python, smoke suite wired 4 layers into core), typeorm/typeorm#12647 (TypeScript, drivers rewired into the platform layer, Qodo contrast), dotnet/efcore#38676 (C#, Query.Internal to root namespace edge seeding a near-cycle, Copilot contrast). Diagrams open zoomed on the flagged component; PR titles deep-link to the original PRs. FAQ: rewritten from a buyer's perspective; adds "how does Striff know my architecture", corrects the unverifiable under-30-seconds claim, and makes the third-party/code-storage answer precise (structural model, not raw source). Extension copy now reflects that a GitHub token is required. Pricing: verified against striff-api's EntitlementTier (5/15/50 caps at $29/$59/$149, 24h upgrade grace); fixes installed.astro claiming 20 repos for Team where the code enforces 15. Header: primary CTA now installs the GitHub App; new amber Chrome-styled Extension button beside it (desktop and mobile menus). Mobile: finding rows and code chips wrap instead of clipping, example cards no longer overflow the viewport, and the gallery diagram opens fit-to-width on phones. Also removes ICCQ references and the last em dashes in shipped HTML. Co-Authored-By: Claude Fable 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01X9W3whWjXLxAQFHY61Gk61

Preserve absolute event timestamps in Gossip

65b9d2f

phanky1 force-pushed the gossip branch from 6ade94a to 65b9d2f Compare July 17, 2026 12:04

auvipy requested a review from Copilot July 18, 2026 09:15

Copilot AI reviewed Jul 18, 2026

Merge branch 'main' into gossip

946534e

auvipy added this to the 5.7.0 milestone Jul 18, 2026

auvipy requested changes Jul 18, 2026

View reviewed changes

Comment thread celery/worker/consumer/gossip.py Resolved

Add unit test

f7b6202

phanky1 requested a review from auvipy July 18, 2026 13:13

Merge branch 'main' into gossip

58526f8

auvipy mentioned this pull request Jul 19, 2026

Clock drift between workers in different timezones #6213

Closed

8 of 11 tasks

phanky1 and others added 3 commits July 20, 2026 02:28

Trigger CI

db75e0e

Merge branch 'main' into gossip

53d1a51

Merge branch 'main' into gossip

bb28259

auvipy approved these changes Jul 27, 2026

View reviewed changes

auvipy merged commit 6e435fc into celery:main Jul 27, 2026
17 checks passed

auvipy added Component: Celerybeat Component: Gossip PR Type: Bugfix labels Jul 27, 2026

phanky1 deleted the gossip branch July 27, 2026 15:08

Navigation Menu

Repository navigation

Preserve absolute event timestamps in Gossip - #10418

Preserve absolute event timestamps in Gossip#10418
auvipy merged 7 commits into
celery:mainfrom
phanky1:gossip

phanky1 commented Jul 17, 2026 •

edited

Loading

Uh oh!

codecov Bot commented Jul 17, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	def Event(type, _fields=None, __dict__=dict, __now__=time.time, **fields):
	"""Create an event.

	Notes:
	An event is simply a dictionary: the only required field is ``type``.
	A ``timestamp`` field will be set to the current time if not provided.
	"""
	event = __dict__(_fields, **fields) if _fields else fields
	if 'timestamp' not in event:
	event.update(timestamp=__now__(), type=type)

	def event_from_message(self, body, localize=True,
	now=time.time, tzfields=_TZGETTER,
	adjust_timestamp=adjust_timestamp,
	CLIENT_CLOCK_SKEW=CLIENT_CLOCK_SKEW):
	type = body['type']
	if type == 'task-sent':
	# clients never sync so cannot use their clock value
	_c = body['clock'] = (self.clock.value or 1) + CLIENT_CLOCK_SKEW
	self.adjust_clock(_c)
	else:
	try:
	clock = body['clock']
	except KeyError:
	body['clock'] = self.forward_clock()
	else:
	self.adjust_clock(clock)

	if localize:
	try:
	offset, timestamp = tzfields(body)
	except KeyError:
	pass
	else:
	body['timestamp'] = adjust_timestamp(timestamp, offset)
	body['local_received'] = now()

Conversation

phanky1 commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Uh oh!

codecov Bot commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

phanky1 commented Jul 17, 2026 •

edited

Loading

codecov Bot commented Jul 17, 2026 •

edited

Loading