Skip to content

Go SDK: fix data race on Edge Worker's activeWorkloads map - #70188

Open
ColtenOuO wants to merge 1 commit into
apache:mainfrom
ColtenOuO:repro-go-sdk-activeworkloads-race
Open

Go SDK: fix data race on Edge Worker's activeWorkloads map#70188
ColtenOuO wants to merge 1 commit into
apache:mainfrom
ColtenOuO:repro-go-sdk-activeworkloads-race

Conversation

@ColtenOuO

Copy link
Copy Markdown
Contributor

Summary

worker.activeWorkloads (a plain map[uuid.UUID]bundlev1.ExecuteTaskWorkload)
is read and written from multiple goroutines with no synchronization:

  • mainLoop() dispatches every fetched job via go w.runWorkload(ctx, ...)
    (a new goroutine per task), and runWorkload writes to the map on start
    and deletes from it on completion.
  • heartbeat() and the drain check in mainLoop() read len(w.activeWorkloads)
    from a separate goroutine.

maxConcurrency defaults to 16, so running multiple tasks concurrently is
the normal operating mode of an Edge Worker, not an edge case. Any time two
tasks start/finish around the same moment a heartbeat or drain check reads
the map's length, this is a data race.

Reproduction

Confirmed two ways before fixing:

  1. go test -race: 4 distinct data races reported, all pointing at the
    documented read/write sites (map assignment, map delete, and the
    unsynchronized len() reads).

  2. Without -race, running the same concurrent access pattern repeatedly
    reliably produces:

    fatal error: concurrent map writes
    
    goroutine 16 [running]:
    internal/runtime/maps.fatal(...)
    github.com/apache/airflow/go-sdk/edge.(*worker).runWorkload(...)
        /go-sdk/edge/worker.go:448
    

    This is a Go runtime fatal error, not a panic -- recover() cannot
    catch it, so it crashes the entire worker process (all in-flight tasks),
    not just the one task being started/finished at that moment.

Changes

  • Add activeWorkloadsMu sync.Mutex to worker, and guard all four
    access points (the write in runWorkload, the delete in its deferred
    cleanup, and the two len() reads in heartbeat() and mainLoop()'s
    drain check).

Was generative AI tooling used to co-author this PR?
  • Yes — Claude Code (Sonnet 5)

Every fetched job is dispatched via `go w.runWorkload(...)`, which reads
and writes worker.activeWorkloads from that goroutine, while heartbeat()
and the drain check in mainLoop() read it from a different goroutine --
all without synchronization. Since maxConcurrency defaults to 16, running
more than one task at a time is the normal case, not an edge case.

Reproduced with `go test -race` (4 distinct races reported) and, without
-race, an actual unrecoverable `fatal error: concurrent map writes` that
crashes the whole worker process, not just one task.

Guard all reads/writes of activeWorkloads with a mutex.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant