1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
|
package main
import (
"fmt"
"log"
"math"
"sync"
"time"
"github.com/emersion/go-imap"
"github.com/emersion/go-imap/client"
)
// ============================================================================
// IMAP connection pool
// ============================================================================
const (
// poolAlpha is the EMA smoothing factor for both pressure and idleness.
// 0.2 ≈ roughly last 5 interactions.
poolAlpha = 0.2
// poolTargetWait is the acceptable wait time to acquire a connection.
// A wait of this duration produces a pressure sample of 1.0.
// 100ms: IMAP round-trips already take ~25ms each, so a 100ms wait
// means the pool is significantly under-provisioned.
poolTargetWait = 100 * time.Millisecond
// poolGrowThreshold: grow when pressure EMA exceeds this.
// 0.3 means: consistently waiting >6ms on average.
poolGrowThreshold = 0.3
// poolShrinkThreshold: shrink when idleness EMA exceeds this.
// 0.7 means: 70%+ of target slots are consistently idle after release.
poolShrinkThreshold = 0.7
)
// poolConn is a single IMAP connection held in the global pool.
// It tracks which mailbox is currently selected so acquire() can
// prefer a connection that is already on the requested mailbox.
type poolConn struct {
c *client.Client
mailbox string // currently selected mailbox
}
// globalPool is a single dynamically-sized pool of IMAP connections shared
// across all mailboxes. Connections carry their currently-selected mailbox;
// acquire(mailbox) prefers an idle connection already on that mailbox to
// avoid a re-SELECT round-trip, but will re-SELECT any idle connection rather
// than open a new one.
//
// Two independent EMA signals drive grow/shrink decisions:
//
// pressure = EMA of min(waitTime / poolTargetWait, 1.0)
// idleness = EMA of (idleCount / size) sampled on each release
//
// pressure measures contention: "how long did callers wait for a slot?"
// idleness measures over-provisioning: "what fraction of the pool is idle?"
//
// Grow by 1 when pressure > 0.3 — callers are waiting too long.
// Shrink by 1 when idleness > 0.7 — 70%+ of target slots are consistently idle.
//
// This separates the two concerns: a burst that uses all connections but
// serves them instantly has high idleness after (→ shrink) but low pressure
// (no waiting) only after the burst ends. During the burst, idleness is low
// (all connections in use) so no premature shrink fires.
//
// Connections are opened lazily — none are created until acquire() is called.
// release() is fire-and-forget: the caller returns immediately and a background
// goroutine performs a NOOP health check, then returns the connection to the
// idle list or discards it. If the pool is shrinking, the connection is
// discarded to permanently reduce the pool size.
type globalPool struct {
creds imapCreds
min int
max int
mu sync.Mutex
size int // current target: min ≤ size ≤ max
outstanding int // connections currently held by callers (for logging)
draining int // slots to permanently discard on next release(s)
pressure float64 // EMA of normalised wait time (grow signal)
idleness float64 // EMA of idle fraction after release (shrink signal)
idle []poolConn // idle ready connections, searched by mailbox on acquire
// sem is a counting semaphore: a slot is claimed by RECEIVING from it,
// and released by SENDING to it. Buffered at max; pre-filled with initSize
// slots. acquire blocks when all slots are taken.
sem chan struct{}
}
// newGlobalPool creates a pool starting at min connections, growing up to max.
// No connections are opened until acquire() is called.
func newGlobalPool(creds imapCreds, min, max int) *globalPool {
initSize := min * 3
if initSize > max {
initSize = max
}
if initSize < 1 {
initSize = 1
}
p := &globalPool{
creds: creds,
min: min,
max: max,
size: initSize,
sem: make(chan struct{}, max),
}
// Pre-fill semaphore with initSize slots.
for i := 0; i < initSize; i++ {
p.sem <- struct{}{}
}
log.Printf("pool: created (min=%d max=%d initSize=%d)", min, max, initSize)
return p
}
func (p *globalPool) logf(format string, args ...any) {
log.Printf("pool: "+format, args...)
}
// popIdle removes and returns an idle connection. Prefers one already
// selected on mailbox; falls back to any idle connection.
// Must be called with p.mu held.
func (p *globalPool) popIdle(mailbox string) (poolConn, bool) {
// First pass: exact mailbox match.
for i, conn := range p.idle {
if conn.mailbox == mailbox {
p.idle[i] = p.idle[len(p.idle)-1]
p.idle = p.idle[:len(p.idle)-1]
return conn, true
}
}
// Second pass: any idle connection (will need re-SELECT).
if len(p.idle) > 0 {
conn := p.idle[len(p.idle)-1]
p.idle = p.idle[:len(p.idle)-1]
return conn, true
}
return poolConn{}, false
}
// acquire blocks until a slot is available, then returns a connection
// SELECT'd on mailbox. It prefers idle connections already on mailbox,
// re-SELECTs any other idle connection, or opens a new one.
//
// The pressure EMA is updated after unblocking, using the actual time spent
// waiting for a slot. This directly captures "how long did callers wait?"
// rather than proxies like utilisation ratio or queue length.
func (p *globalPool) acquire(mailbox string) (*client.Client, error) {
// Measure how long we actually block waiting for a slot.
start := time.Now()
<-p.sem // blocks until a slot is available
waited := time.Since(start)
p.mu.Lock()
p.outstanding++
// Update pressure EMA using log-scaled wait time.
// log(1 + waitMs) / log(1 + targetMs) maps:
// 0ms → 0.0 (instant)
// targetMs → 1.0
// >>targetMs → >1.0 (capped at 1.0)
// The log scale means the difference between 0ms and 1ms matters more
// than the difference between 900ms and 1000ms.
waitMs := waited.Seconds() * 1000
targetMs := poolTargetWait.Seconds() * 1000
sample := min(math.Log1p(waitMs)/math.Log1p(targetMs), 1.0)
p.pressure = poolAlpha*sample + (1-poolAlpha)*p.pressure
// Grow if callers are waiting too long and pool is below max.
// Reset idleness to 0 on grow: a grow event means the pool was
// under-provisioned, so the old idleness history is stale and
// would otherwise cause an immediate shrink after the burst ends.
if p.pressure > poolGrowThreshold && p.size < p.max {
p.size++
p.idleness = 0
p.logf("grow size=%d pressure=%.2f waited=%s", p.size, p.pressure, waited)
select {
case p.sem <- struct{}{}:
default:
}
}
// Loop to skip over stale idle connections. We already hold our semaphore
// slot, so we never return it inside the loop — just try the next idle
// connection or fall through to opening a new one.
for {
conn, hasIdle := p.popIdle(mailbox)
outstanding := p.outstanding
idleCount := len(p.idle)
p.mu.Unlock()
if hasIdle {
// Discard connections the server has already closed on us.
if conn.c.State() == imap.LogoutState {
conn.c.Logout()
p.logf("acquire: idle connection already closed, discarding[%s]", conn.mailbox)
p.mu.Lock()
// outstanding stays the same — we still hold the slot and
// will hand it to the next connection we find/open.
continue
}
if conn.mailbox == mailbox {
// Exact match — reuse without any extra round-trip.
p.logf("acquire reuse[%s] (size=%d outstanding=%d idle=%d pressure=%.2f idleness=%.2f waited=%s)",
mailbox, p.size, outstanding, idleCount, p.pressure, p.idleness, waited)
return conn.c, nil
}
// Re-SELECT to switch mailbox — one round-trip, cheaper than a new connection.
p.logf("acquire reselect[%s→%s] (size=%d outstanding=%d idle=%d pressure=%.2f idleness=%.2f waited=%s)",
conn.mailbox, mailbox, p.size, outstanding, idleCount, p.pressure, p.idleness, waited)
if _, err := conn.c.Select(mailbox, true); err != nil {
// Connection is broken — discard and try the next idle one or open fresh.
conn.c.Logout()
p.logf("acquire reselect[%s→%s]: connection broken, discarding: %v", conn.mailbox, mailbox, err)
p.mu.Lock()
continue
}
return conn.c, nil
}
// No idle connection — open a new one.
p.logf("acquire new[%s] (size=%d outstanding=%d idle=%d pressure=%.2f idleness=%.2f waited=%s)",
mailbox, p.size, outstanding, idleCount, p.pressure, p.idleness, waited)
c, err := connectIMAP(p.creds)
if err != nil {
p.mu.Lock()
p.outstanding--
p.mu.Unlock()
p.sem <- struct{}{} // return the slot
return nil, fmt.Errorf("pool open: %w", err)
}
if _, err := c.Select(mailbox, true); err != nil {
c.Logout()
p.mu.Lock()
p.outstanding--
p.mu.Unlock()
p.sem <- struct{}{} // return the slot
return nil, fmt.Errorf("pool select %s: %w", mailbox, err)
}
return c, nil
}
}
// release returns a connection to the pool asynchronously.
// The caller must pass the mailbox the connection is currently selected on.
// The caller returns immediately; a background goroutine:
// 1. Performs a NOOP health check.
// 2. Decrements outstanding.
// 3. If the pool is shrinking (draining > 0) or the connection is unhealthy,
// discards the connection WITHOUT returning the slot to sem — permanently
// shrinking the pool by 1.
// 4. Otherwise returns the connection to the idle list and sends the slot back to sem.
func (p *globalPool) release(c *client.Client, mailbox string) {
go func() {
// Health check first — before touching any pool state.
healthy := true
if err := c.Noop(); err != nil {
p.logf("release: connection unhealthy, discarding: %v", err)
c.Logout()
healthy = false
}
p.mu.Lock()
p.outstanding--
shouldDrain := p.draining > 0
if shouldDrain {
p.draining--
}
p.mu.Unlock()
if !healthy {
// Unhealthy connection — discard it but return the slot to sem
// so the pool can replenish. The pool size is unchanged.
p.logf("release: unhealthy, slot returned (size=%d)", p.size)
p.sem <- struct{}{}
return
}
if shouldDrain {
// Deliberate shrink — consume the slot permanently.
p.logf("release: drained slot (size=%d)", p.size)
c.Logout()
return
}
// Return connection to idle list for reuse.
p.mu.Lock()
p.idle = append(p.idle, poolConn{c: c, mailbox: mailbox})
idleCount := len(p.idle)
// Sample idleness AFTER returning the connection: fraction of
// target slots that are idle. Using size (not outstanding+idle)
// means unused lazy slots count against idleness — if size=8 but
// only 4 connections were ever opened, idleness stays low and we
// won't shrink the target below what we've actually used.
idleSample := float64(idleCount) / float64(max(p.size, 1))
p.idleness = poolAlpha*idleSample + (1-poolAlpha)*p.idleness
// Shrink if the pool is consistently mostly idle and above min,
// but only if there is actually an open idle connection to close —
// no point draining a lazy slot that was never opened.
// Reset idleness to 0 on shrink so we re-observe behaviour before
// deciding to shrink again.
if p.idleness > poolShrinkThreshold && p.size > p.min && p.draining == 0 && idleCount > 0 {
p.size--
p.idleness = 0
p.draining++
p.logf("shrink size=%d", p.size)
}
p.logf("release: returned[%s] (size=%d outstanding=%d idle=%d pressure=%.2f idleness=%.2f)",
mailbox, p.size, p.outstanding, idleCount, p.pressure, p.idleness)
p.mu.Unlock()
// Return the slot to sem so the next acquire can proceed.
p.sem <- struct{}{}
}()
}
// close logs out all idle connections. Outstanding connections are not
// affected — they will be discarded when released after close.
func (p *globalPool) close() {
p.mu.Lock()
idle := p.idle
p.idle = nil
p.mu.Unlock()
for _, conn := range idle {
conn.c.Logout()
}
}
|