golang.org/x/sync v0.20.0: errgroup.Group, errgroup.WithContext, singleflight.Group

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify pkg:golang/golang.org/x/sync@v0.20.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:golang/golang.org/x/sync@v0.20.0
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - runtime: node@22.23.2

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

package sample

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"golang.org/x/sync/errgroup"
)

func TestErrgroupWithContextAndCancellation(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	runner, gctx := NewTaskRunner(ctx)
	targetErr := errors.New("simulated failure")

	var task2ObservedCancel int32

	runner.Go(func() error {
		return targetErr
	})

	runner.Go(func() error {
		select {
		case <-gctx.Done():
			atomic.StoreInt32(&task2ObservedCancel, 1)
			return nil
		case <-time.After(2 * time.Second):
			return errors.New("timeout waiting for context cancellation")
		}
	})

	err := runner.Wait()
	if !errors.Is(err, targetErr) {
		t.Fatalf("expected error %v, got %v", targetErr, err)
	}

	if atomic.LoadInt32(&task2ObservedCancel) != 1 {
		t.Fatal("expected second task to observe context cancellation")
	}
}

func TestErrgroupSetLimitAndTryGo(t *testing.T) {
	var g errgroup.Group
	g.SetLimit(1)

	started := make(chan struct{})
	unblock := make(chan struct{})

	// Submit first task filling the limit
	ok1 := g.TryGo(func() error {
		close(started)
		<-unblock
		return nil
	})
	if !ok1 {
		t.Fatal("expected first TryGo to succeed")
	}

	<-started

	// Second task should fail TryGo immediately because limit is 1 and 1 task is running
	ok2 := g.TryGo(func() error {
		return nil
	})
	if ok2 {
		t.Fatal("expected second TryGo to return false when limit is reached")
	}

	close(unblock)
	if err := g.Wait(); err != nil {
		t.Fatalf("unexpected error waiting for group: %v", err)
	}
}

func TestSingleflightDuplicateSuppression(t *testing.T) {
	dedup := NewDeduplicator()
	key := "shared-key"

	var executionCount int64
	started := make(chan struct{})
	unblock := make(chan struct{})

	var wg sync.WaitGroup
	wg.Add(2)

	var res1, res2 any
	var err1, err2 error
	var shared1, shared2 bool

	go func() {
		defer wg.Done()
		res1, err1, shared1 = dedup.Do(key, func() (any, error) {
			atomic.AddInt64(&executionCount, 1)
			close(started)
			<-unblock
			return "computed-data", nil
		})
	}()

	go func() {
		defer wg.Done()
		<-started
		res2, err2, shared2 = dedup.Do(key, func() (any, error) {
			atomic.AddInt64(&executionCount, 1)
			return "unexpected-data", nil
		})
	}()

	<-started
	time.Sleep(10 * time.Millisecond)
	close(unblock)
	wg.Wait()

	if err1 != nil || err2 != nil {
		t.Fatalf("unexpected errors: err1=%v, err2=%v", err1, err2)
	}
	if res1 != "computed-data" || res2 != "computed-data" {
		t.Fatalf("expected identical result, got res1=%v, res2=%v", res1, res2)
	}
	if !shared1 && !shared2 {
		t.Fatal("expected at least one caller to observe shared=true")
	}
	if count := atomic.LoadInt64(&executionCount); count != 1 {
		t.Fatalf("expected exactly 1 execution, got %d", count)
	}
}

func TestSingleflightForget(t *testing.T) {
	dedup := NewDeduplicator()
	key := "forget-key"

	var executionCount int64

	v1, err1, _ := dedup.Do(key, func() (any, error) {
		atomic.AddInt64(&executionCount, 1)
		return "data-1", nil
	})
	if err1 != nil || v1 != "data-1" {
		t.Fatalf("unexpected result: %v, err: %v", v1, err1)
	}

	dedup.Forget(key)

	v2, err2, shared := dedup.Do(key, func() (any, error) {
		atomic.AddInt64(&executionCount, 1)
		return "data-2", nil
	})
	if err2 != nil || v2 != "data-2" {
		t.Fatalf("unexpected result: %v, err: %v", v2, err2)
	}
	if shared {
		t.Fatal("expected shared to be false after Forget")
	}
	if count := atomic.LoadInt64(&executionCount); count != 2 {
		t.Fatalf("expected 2 executions after forget, got %d", count)
	}
}

func TestSemaphoreWeightedAcquireAndTryAcquire(t *testing.T) {
	ctx := context.Background()
	limiter := NewResourceLimiter(3)

	if !limiter.TryAcquire(2) {
		t.Fatal("expected TryAcquire(2) to succeed on limiter with capacity 3")
	}
	if !limiter.TryAcquire(1) {
		t.Fatal("expected TryAcquire(1) to succeed for remaining capacity 1")
	}
	if limiter.TryAcquire(1) {
		t.Fatal("expected TryAcquire(1) to fail when capacity is exhausted")
	}

	limiter.Release(3)

	if err := limiter.Acquire(ctx, 3); err != nil {
		t.Fatalf("unexpected error on Acquire: %v", err)
	}
	limiter.Release(3)
}

{"case":{"caseId":"case:sha256:16a0d4ccd7e594417776b34c7a8982fa1f430dcf5726ffe4f9e958295d7f5ea8","contract":["errgroup.WithContext cancels derived context upon task error and Wait returns the first error","errgroup.Group with SetLimit constrains concurrency and TryGo returns false when capacity is reached","singleflight.Group suppresses duplicate concurrent function executions for identical keys and shares result","singleflight.Group Forget removes key allowing subsequent calls to execute afresh","semaphore.Weighted bounds concurrent resource access with Acquire, Release, and TryAcquire"],"goal":"verify pkg:golang/golang.org/x/sync@v0.20.0","kind":"HOW","packages":["pkg:golang/golang.org/x/sync@v0.20.0"],"schemaVersion":1,"symbols":["errgroup.Group","errgroup.WithContext","singleflight.Group","semaphore.Weighted"]},"contractCommand":["go","test","./..."],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"golang","language":"go","libc":"glibc","libcVersion":"2.39","moduleSystem":"go.mod","os":"linux","osVersionBucket":"24","packageManager":"go","runtime":"go","runtimeVersion":"1.26","schemaVersion":1},"license":"MIT-0","packages":["pkg:golang/golang.org/x/sync@v0.20.0"],"schemaVersion":1,"subject":"pkg:golang/golang.org/x/sync@v0.20.0","symbols":["errgroup.Group","errgroup.WithContext","singleflight.Group","semaphore.Weighted"],"verifierAdapter":"golang@1"}

module sample

go 1.25.0

require golang.org/x/sync v0.20.0

golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=

{
  "schemaVersion": 1,
  "goal": "verify pkg:golang/golang.org/x/sync@v0.20.0",
  "kind": "HOW",
  "packages": [
    "pkg:golang/golang.org/x/sync@v0.20.0"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "runtime": "node@22.23.2"
  }
}

package sample

import (
	"context"

	"golang.org/x/sync/errgroup"
	"golang.org/x/sync/semaphore"
	"golang.org/x/sync/singleflight"
)

// TaskRunner coordinates concurrent tasks with bounded concurrency or error cancellation.
type TaskRunner struct {
	group *errgroup.Group
	ctx   context.Context
}

// NewTaskRunner creates a TaskRunner with an errgroup context.
func NewTaskRunner(ctx context.Context) (*TaskRunner, context.Context) {
	g, gctx := errgroup.WithContext(ctx)
	return &TaskRunner{
		group: g,
		ctx:   gctx,
	}, gctx
}

// Go executes a function within the errgroup.
func (r *TaskRunner) Go(f func() error) {
	r.group.Go(f)
}

// Wait waits for all tasks in the runner to complete and returns the first error.
func (r *TaskRunner) Wait() error {
	return r.group.Wait()
}

// Deduplicator suppresses duplicate function calls for identical keys.
type Deduplicator struct {
	sf singleflight.Group
}

// NewDeduplicator creates a new Deduplicator instance.
func NewDeduplicator() *Deduplicator {
	return &Deduplicator{}
}

// Do executes the function or shares the result if already in flight.
func (d *Deduplicator) Do(key string, fn func() (any, error)) (any, error, bool) {
	return d.sf.Do(key, fn)
}

// Forget removes the key from in-flight tracking.
func (d *Deduplicator) Forget(key string) {
	d.sf.Forget(key)
}

// ResourceLimiter wraps a weighted semaphore for resource capacity control.
type ResourceLimiter struct {
	sem *semaphore.Weighted
}

// NewResourceLimiter creates a new limiter with total capacity n.
func NewResourceLimiter(n int64) *ResourceLimiter {
	return &ResourceLimiter{
		sem: semaphore.NewWeighted(n),
	}
}

// Acquire acquires n units, blocking until available or context canceled.
func (l *ResourceLimiter) Acquire(ctx context.Context, n int64) error {
	return l.sem.Acquire(ctx, n)
}

// TryAcquire attempts to acquire n units without blocking.
func (l *ResourceLimiter) TryAcquire(n int64) bool {
	return l.sem.TryAcquire(n)
}

// Release releases n units back to the limiter.
func (l *ResourceLimiter) Release(n int64) {
	l.sem.Release(n)
}