1package app
2
3import (
4 "context"
5 "fmt"
6
7 "github.com/bnema/gordon/internal/boundaries/out"
8 "github.com/bnema/gordon/internal/usecase/apptraffic"
9)
10
11// appTrafficPublisher is the single serialized HTTP/L4 publication
12// boundary for apps. Activation rebuilds, fail-closed recovery
13// withdrawal, and config reload all take the same mutex: the host index
14// replacement and the runtime graph application always run as one
15// operation, so no caller can observe an index that disagrees with the
16// applied graph, and a stale config can never win a race.
17type appTrafficPublisher struct {
18 // gate is a one-slot semaphore instead of a mutex so a waiting
19 // caller can abandon the wait when its context ends: a monitor pass
20 // must never hold an app lock for a full graph application that a
21 // sibling publication started, and shutdown must not block behind
22 // it.
23 gate chan struct{}
24 svc *services
25 cfg Config
26 // applyGraph applies one prepared configuration. The publication
27 // projects and validates a candidate against this call and swaps the
28 // live host index only after it returns nil, so a rejected graph can
29 // never leave the index ahead of the dataplane. It is a field so
30 // publication ordering is testable without a live traffic manager.
31 applyGraph func(ctx context.Context, cfg Config, hosts appHostRoutes) error
32}
33
34var _ out.AppTrafficRefresher = (*appTrafficPublisher)(nil)
35
36// newAppTrafficPublisher wires the boundary. The installation config is
37// stored so activation rebuilds use the same validated snapshot that
38// reload applied.
39func newAppTrafficPublisher(svc *services, cfg Config) *appTrafficPublisher {
40 publisher := &appTrafficPublisher{gate: make(chan struct{}, 1), svc: svc, cfg: cfg}
41 publisher.applyGraph = func(ctx context.Context, cfg Config, hosts appHostRoutes) error {
42 return applyTrafficRuntimeConfig(ctx, svc.trafficManager, cfg, svc.configSvc, hosts)
43 }
44 publisher.gate <- struct{}{}
45 return publisher
46}
47
48// acquire waits for the publication gate or the caller's deadline.
49func (p *appTrafficPublisher) acquire(ctx context.Context) error {
50 select {
51 case <-ctx.Done():
52 return ctx.Err()
53 case <-p.gate:
54 return nil
55 }
56}
57
58// release returns the publication gate to the next caller.
59func (p *appTrafficPublisher) release() {
60 p.gate <- struct{}{}
61}
62
63// RebuildTraffic re-projects verified ACTIVE state and applies the full
64// HTTP/L4 graph.
65func (p *appTrafficPublisher) RebuildTraffic(ctx context.Context) error {
66 if err := p.acquire(ctx); err != nil {
67 return fmt.Errorf("traffic publisher: %w", err)
68 }
69 defer p.release()
70 return p.rebuild(ctx, p.cfg)
71}
72
73// RebuildWithConfig applies the current validated config through the
74// same serialization point. Startup and reload call it so they cannot
75// interleave with a recovery withdrawal.
76func (p *appTrafficPublisher) RebuildWithConfig(ctx context.Context, cfg Config) error {
77 if err := p.acquire(ctx); err != nil {
78 return fmt.Errorf("traffic publisher: %w", err)
79 }
80 defer p.release()
81 p.cfg = cfg
82 return p.rebuild(ctx, cfg)
83}
84
85// WithdrawService makes one service non-forwardable and republishes.
86// The durable ACTIVE record loses the service's binds first: every
87// later projection is then fail-closed even if the graph application
88// itself fails. A non-nil result means stale forwarding may remain and
89// the caller must retain its publication inhibition.
90// WithdrawServiceState is the canonical state-only ACTIVE bind
91// withdrawal: it clears the recorded binds through the same
92// serialization point as WithdrawService but applies no graph.
93func (p *appTrafficPublisher) WithdrawServiceState(ctx context.Context, app, service string) error {
94 if err := p.acquire(ctx); err != nil {
95 return fmt.Errorf("traffic publisher: %w", err)
96 }
97 defer p.release()
98 return p.withdrawFromState(ctx, app, service)
99}
100
101func (p *appTrafficPublisher) WithdrawService(ctx context.Context, app, service string) error {
102 if err := p.acquire(ctx); err != nil {
103 return fmt.Errorf("traffic publisher: %w", err)
104 }
105 defer p.release()
106 if err := p.withdrawFromState(ctx, app, service); err != nil {
107 return err
108 }
109 err := p.rebuild(ctx, p.cfg)
110 if err == nil {
111 return nil
112 }
113 // The graph apply failed while the durable ACTIVE record already
114 // dropped the service's binds. The read model still has to follow: a
115 // stale index entry could dial a loopback port that has been
116 // recycled by another workload, so the proxy must fail closed. The
117 // error is returned unchanged so the caller keeps its publication
118 // inhibition and retries the graph.
119 if candidate, projErr := p.projectCandidate(ctx, p.cfg); projErr == nil && candidate != nil {
120 p.svc.appHostIndex.Replace(candidate.Entries())
121 }
122 return err
123}
124
125// withdrawFromState drops the recorded binds of one service in ACTIVE.
126// The container itself is untouched: withdrawal is a traffic decision,
127// never a workload mutation.
128func (p *appTrafficPublisher) withdrawFromState(ctx context.Context, app, service string) error {
129 if p.svc.appState == nil {
130 return nil
131 }
132 active, ok, err := p.svc.appState.LoadActive(ctx, app)
133 if err != nil {
134 return fmt.Errorf("withdraw %s/%s: load active: %w", app, service, err)
135 }
136 if !ok {
137 return nil
138 }
139 current, ok := active.Services[service]
140 if !ok || (len(current.BackendBinds) == 0 && len(current.UDPBackendBinds) == 0) {
141 return nil
142 }
143 current.BackendBinds = nil
144 current.UDPBackendBinds = nil
145 active.Services[service] = current
146 if err := p.svc.appState.SaveActive(ctx, active); err != nil {
147 return fmt.Errorf("withdraw %s/%s: persist fail-closed state: %w", app, service, err)
148 }
149 return nil
150}
151
152// rebuild projects ACTIVE into a candidate host view, applies the full
153// graph against that candidate, and publishes the candidate only after
154// the dataplane accepted it. The caller holds the publication gate. On
155// any failure the live index and the applied graph both stay at their
156// previous generation: the proxy never serves routing the dataplane
157// rejected, and no retirement ever follows an unpublished switch.
158func (p *appTrafficPublisher) rebuild(ctx context.Context, cfg Config) error {
159 candidate, err := p.projectCandidate(ctx, cfg)
160 if err != nil {
161 return err
162 }
163 var hosts appHostRoutes
164 if candidate != nil {
165 hosts = candidate
166 }
167 if err := p.applyGraph(ctx, cfg, hosts); err != nil {
168 return fmt.Errorf("apply traffic graph: %w", err)
169 }
170 if candidate != nil {
171 p.svc.appHostIndex.Replace(candidate.Entries())
172 }
173 return nil
174}
175
176// projectCandidate builds the detached host view for one publication.
177// The candidate never touches the live index: a failing projection leaves
178// what the proxy serves untouched. It returns nil when app state is not
179// wired, in which case the graph carries configuration routes only.
180func (p *appTrafficPublisher) projectCandidate(ctx context.Context, cfg Config) (*apptraffic.HostIndex, error) {
181 if p.svc.appActivator == nil || p.svc.appState == nil || p.svc.appHostIndex == nil {
182 return nil, nil
183 }
184 candidate := apptraffic.NewHostIndex()
185 if err := p.svc.appActivator.RebuildHostIndex(ctx, candidate, p.svc.appState, appEntrypointPolicies(cfg)); err != nil {
186 return nil, fmt.Errorf("project app host index: %w", err)
187 }
188 return candidate, nil
189}