struxadotcloud/struxa

public

initial commitd4cb9e15 months ago158 lines · 5.0 KB

1	# Drain Pipeline Reference
2
3	The drain pipeline wraps any adapter to add batching, retry with backoff, and buffer overflow protection. Use it in production to reduce network overhead and handle transient failures.
4
5	## When to Recommend
6
7	- The user has a drain adapter (Axiom, OTLP, custom) and is deploying to production
8	- High-throughput scenarios where one HTTP request per event is wasteful
9	- The user needs retry logic for unreliable backends
10	- The user is implementing batching manually with `setInterval` and arrays
11
12	## Basic Setup
13
14	```typescript
15	// server/plugins/evlog-drain.ts
16	import type { DrainContext
17	import { createDrainPipeline }
18	import { createAxiomDrain }
19
20	export default defineNitroPlugin((nitroApp
21	const pipeline =
22	const drain =
23
24	nitroApp.hooks.hook
25	nitroApp.hooks.hook
26	});
27	```
28
29	Important: Always call `drain.flush()
30
31	## Full Configuration
32
33	```typescript
34	const pipeline = createDrainPipeline
35	batch: {
36	size: 50, //
37	intervalMs: 5000, //
38	},
39	retry: {
40	maxAttempts: 3, //
41	backoff: "exponential", //
42	initialDelayMs: 1000, //
43	maxDelayMs: 30000, //
44	},
45	maxBufferSize: 1000, //
46	onDropped: (events,
47	// Called when events are dropped (overflow or retry exhaustion)
48	console.error(`[evlog] Dropped
49	},
50	});
51	```
52
53	## How It Works
54
55	1. `drain(ctx)` pushes a single event into the buffer
56	2. When `buffer.length >= batch.size`, the batch is flushed immediately
57	3. If the batch isn't full, a timer starts; after `intervalMs`, whatever is buffered gets flushed
58	4. On flush, the drain function receives `T[]` (always an array)
59	5. If the drain throws, the batch is retried with the configured backoff
60	6. After `maxAttempts` failures, `onDropped` is called and the batch is discarded
61	7. If the buffer exceeds `maxBufferSize`, the oldest event is dropped and `onDropped` is called
62
63	## Backoff Strategies
64
65	\| Strategy \| Delay Pattern \| Best For \|
66	\| ------------- \| ----------------- \|
67	\| `exponential` \| 1s, 2s, 4s, 8s... \| Default. Transient failures needing recovery time
68	\| `linear` \| 1s, 2s, 3s, 4s... \| Predictable delay growth
69	\| `fixed` \| 1s, 1s, 1s, 1s... \| Rate-limited APIs with known cooldown
70
71	## Returned Drain Function API
72
73	```typescript
74	const drain = pipeline
75
76	drain(ctx); //
77	await drain.flush()
78	drain.pending; //
79	```
80
81	## Common Patterns
82
83	### With multiple adapters
84
85	```typescript
86	const axiom = createAxiomDrain
87	const otlp = createOTLPDrain
88
89	const pipeline = createDrainPipeline
90	const drain = pipeline
91	await Promise.allSettled
92	});
93	```
94
95	### Custom drain function
96
97	```typescript
98	const pipeline = createDrainPipeline
99	const drain = pipeline
100	await fetch("https://your-service.com/logs
101	method: "POST",
102	headers: { "Content-Type":
103	body: JSON.stringify
104	});
105	});
106	```
107
108	### Low-traffic with longer interval
109
110	```typescript
111	const pipeline = createDrainPipeline
112	batch: { size: 10
113	});
114	```
115
116	## Anti-Patterns
117
118	### Manual batching with setInterval
119
120	```typescript
121	// ❌ No retry, no overflow protection, no flush on shutdown
122	const batch: WideEvent
123	setInterval(() => {
124	if (batch.length
125	}, 5000)
126	```
127
128	Transform to:
129
130	```typescript
131	// ✅ Use the pipeline
132	const pipeline = createDrainPipeline
133	const drain = pipeline
134	nitroApp.hooks.hook(
135	```
136
137	### Missing flush on shutdown
138
139	```typescript
140	// ❌ Buffered events lost on process exit
141	nitroApp.hooks.hook(
142	```
143
144	Fix:
145
146	```typescript
147	// ✅ Always flush on close
148	nitroApp.hooks.hook(
149	nitroApp.hooks.hook(
150	```
151
152	## Review Checklist
153
154	- [ ] Pipeline wraps the adapter for production use
155	- [ ] `drain.flush()` called on server `close` hook
156	- [ ] `onDropped` callback logs or reports dropped events
157	- [ ] Batch size and interval are appropriate for the traffic volume
158	- [ ] `maxBufferSize` is set to prevent memory leaks under load