coordinator.go RawHistoryFolder

1package deployment
2
3import (
4	"context"
5	"sync"
6)
7
8// appCoordinator serializes workload mutations per app. User mutations
9// and boot acquire the app lock blocking; periodic reconciliation
10// probes it with TryAcquire and skips busy apps instead of waiting.
11// Entries are never deleted so a lock cannot be re-created under a
12// holder; the map stays bounded by known app names.
13type appCoordinator struct {
14	mu    sync.Mutex
15	locks map[string]chan struct{}
16}
17
18func newAppCoordinator() *appCoordinator {
19	return &appCoordinator{locks: map[string]chan struct{}{}}
20}
21
22// lockFor returns the app's lock channel, creating it free (token
23// present) on first use. The caller must hold mu.
24func (c *appCoordinator) lockFor(app string) chan struct{} {
25	ch, ok := c.locks[app]
26	if !ok {
27		ch = make(chan struct{}, 1)
28		ch <- struct{}{}
29		c.locks[app] = ch
30	}
31	return ch
32}
33
34// acquire blocks until the app lock is held or ctx ends. The returned
35// release must be called exactly once by the holder.
36func (c *appCoordinator) acquire(ctx context.Context, app string) (func(), error) {
37	c.mu.Lock()
38	ch := c.lockFor(app)
39	c.mu.Unlock()
40	select {
41	case <-ctx.Done():
42		return nil, ctx.Err()
43	case <-ch:
44		return func() { ch <- struct{}{} }, nil
45	}
46}
47
48// tryAcquire takes the app lock without blocking. ok is false when the
49// app is busy; the caller must skip the app in that case.
50func (c *appCoordinator) tryAcquire(app string) (release func(), ok bool) {
51	c.mu.Lock()
52	ch := c.lockFor(app)
53	c.mu.Unlock()
54	select {
55	case <-ch:
56		return func() { ch <- struct{}{} }, true
57	default:
58		return nil, false
59	}
60}