| 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 |