diff --git a/apps/edge/internal/service/run_wire.go b/apps/edge/internal/service/run_wire.go index 6bf07a41..badb07ce 100644 --- a/apps/edge/internal/service/run_wire.go +++ b/apps/edge/internal/service/run_wire.go @@ -2,6 +2,7 @@ package service import ( "fmt" + "sync/atomic" "time" "google.golang.org/protobuf/types/known/structpb" @@ -10,8 +11,23 @@ import ( iop "iop/proto/gen/iop" ) +var lastRunIDNanos atomic.Int64 + func NewRunID() string { - return fmt.Sprintf("manual-%d", time.Now().UnixNano()) + return newRunIDAt(time.Now().UnixNano()) +} + +func newRunIDAt(now int64) string { + for { + previous := lastRunIDNanos.Load() + next := now + if next <= previous { + next = previous + 1 + } + if lastRunIDNanos.CompareAndSwap(previous, next) { + return fmt.Sprintf("manual-%d", next) + } + } } func IsNodeDisconnected(event *iop.EdgeNodeEvent) bool { diff --git a/apps/edge/internal/service/service_internal_test.go b/apps/edge/internal/service/service_internal_test.go index ee0ad705..838afbb4 100644 --- a/apps/edge/internal/service/service_internal_test.go +++ b/apps/edge/internal/service/service_internal_test.go @@ -18,6 +18,32 @@ import ( iop "iop/proto/gen/iop" ) +func TestRunIDRemainsUniqueAtSameClockTick(t *testing.T) { + const count = 128 + ids := make(chan string, count) + var wg sync.WaitGroup + wg.Add(count) + for i := 0; i < count; i++ { + go func() { + defer wg.Done() + ids <- newRunIDAt(1) + }() + } + wg.Wait() + close(ids) + + seen := make(map[string]struct{}, count) + for id := range ids { + if _, exists := seen[id]; exists { + t.Fatalf("duplicate run id %q", id) + } + seen[id] = struct{}{} + } + if len(seen) != count { + t.Fatalf("unique run ids = %d, want %d", len(seen), count) + } +} + // TestRefreshProviderCapacityAffectsNextDispatch verifies that a provider // capacity change applied via SetRuntimeConfig is reflected in the candidate // capacity used for the next dispatch admission, since candidates are rebuilt