ตัวหมุนโหลด (loading spinner) ไม่ได้บอกอะไรคุณเลย เมื่อต้องรอกระบวนการ AI ที่กินเวลานานหลายนาที หรือต้องกลับไปเข้าคิวเพื่อลองใหม่เป็นครั้งที่สาม คุณจำเป็นต้องเห็นสถานะที่เกิดขึ้นจริง Server-Sent Events (SSE) ช่วยให้คุณมองเห็นสิ่งนั้นได้โดยไม่ต้องมี overhead ของการทำ handshake แบบ WebSockets หรือความยุ่งยากในการจัดการ long polling โดยเซิร์ฟเวอร์จะเปิด HTTP response ทิ้งไว้เพียงหนึ่งเดียวและคอยส่งข้อมูลอัปเดตแบบ plain-text เมื่อมีการเปลี่ยนแปลง ส่วนไคลเอนต์ก็จะอ่านข้อมูลเหล่านั้นทันทีที่ได้รับ

หากการเชื่อมต่อหลุด คุณคงไม่อยากเริ่มนับหนึ่งใหม่ SSE stream ที่สร้างมาอย่างดีจะจดจำได้ว่าคุณค้างอยู่ที่จุดไหน ด้วย Node.js 20 และ standard library เพียงอย่างเดียว คุณก็สามารถเชื่อมต่อระบบนี้ได้แล้ว โดยไม่จำเป็นต้องใช้แพ็กเกจภายนอกเลย

รูปแบบข้อมูลบนสายส่ง (wire format) เป็นอย่างไร

ข้อความ SSE คือข้อความธรรมดา เซิร์ฟเวอร์จะเขียนข้อมูลสามอย่าง ได้แก่ ชื่อเหตุการณ์ (event name) ซึ่งจะมีหรือไม่มีก็ได้, ฟิลด์ data ที่จำเป็นต้องมี และฟิลด์ id ซึ่งจะทำหน้าที่เป็นจุดบันทึก (save point) ของคุณ แต่ละเรคคอร์ดจะจบด้วยอักขระขึ้นบรรทัดใหม่สองตัว (two newline characters) ซึ่งก็คือบรรทัดว่างที่ใช้ระบุขอบเขตของข้อมูล

สตรีมที่ทำงานปกติอาจมีหน้าตาบนสายส่งดังนี้:

id: 14
event: status
data: {"phase":"testing","progress":43}

id: 15
event: status
data: {"phase":"retrying","attempt":2}

ไคลเอนต์ EventSource ของเบราว์เซอร์จะอ่านบรรทัดเหล่านี้โดยอัตโนมัติ โดยจะสร้าง event ขึ้นมาสำหรับแต่ละบล็อกและเก็บ id ล่าสุดไว้ภายใน หากการเชื่อมต่อ TCP หลุด ไคลเอนต์จะรอและทำการเชื่อมต่อใหม่ พร้อมกับส่ง identifier ที่เก็บไว้กลับไปยังเซิร์ฟเวอร์ผ่าน header Last-Event-ID ซึ่ง header ตัวนี้คือเหตุผลสำคัญที่ทำให้รูปแบบนี้ใช้งานได้จริง หากไม่มีมัน คุณจะไม่มีเคอร์เซอร์ที่คงทน (durable cursor) เลย

การเชื่อมต่อเซิร์ฟเวอร์ใน Node.js

โมดูล http ที่มากับ Node สามารถจัดการเรื่องนี้ได้โดยตรง เมื่อมี request เข้ามา ให้ตั้งค่า header ให้ถูกต้องเพื่อให้ไคลเอนต์ทราบว่านี่คือสตรีม ไม่ใช่หน้าเว็บ:

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

ยกเลิกการทำ buffering เนื่องจาก proxy และ framework บางตัวอาจทำการรวบรวม response ไว้เป็นชุด (batch) ซึ่งจะทำลายความรู้สึกแบบ real-time ดังนั้นควรทำการ flush ข้อมูลหลังจากส่งแต่ละ chunk

ส่ง ID ก่อน ตามด้วยประเภทของ event จากนั้นจึงเป็นข้อมูล payload และปิดท้ายด้วยบรรทัดว่าง ลำดับมีความสำคัญเพียงแค่ ID ต้องมาถึงก่อนบรรทัดว่างเพื่อให้ไคลเอนต์สามารถบันทึกมันได้ หากคุณใช้ response.write() แบบ native ผลลัพธ์ที่ได้จะเป็นดังนี้:

response.write(`id: ${cursor}\n`);
response.write(`event: ${eventName}\n`);
response.write(`data: ${JSON.stringify(payload)}\n\n`);

\n\n ที่อยู่ท้ายสุดนั้นไม่ใช่แค่การตกแต่ง แต่ SSE parser จะถือว่ามันคือตัวปิดท้ายเรคคอร์ด (record terminator) หากคุณลืมใส่ ไคลเอนต์จะค้างเพื่อรอข้อมูลเพิ่มเติม

เคอร์เซอร์คือหัวใจสำคัญ

การเชื่อมต่อ HTTP ใหม่ไม่ได้การันตีว่าสถานะจะเริ่มใหม่เสมอไป เมื่อไคลเอนต์เชื่อมต่อใหม่ header Last-Event-ID จะบอกคุณว่าข้อความล่าสุดที่พวกเขาได้รับคืออะไร หน้าที่ของคุณคือการเริ่มส่งต่อจากข้อความถัดไป ไม่ใช่เริ่มใหม่ตั้งแต่ต้น

สิ่งนี้หมายถึงการรักษา log หรือ journal ของเหตุการณ์ที่เรียงลำดับไว้ทางฝั่งเซิร์ฟเวอร์ การใช้ in-memory array อาจใช้ได้สำหรับการสาธิต แต่ในระบบ production คุณควรใช้สิ่งที่คงทนกว่านั้น เช่น การบันทึกลงใน database log, Redis stream หรือ write-ahead journal เพราะการรีสตาร์ทเซิร์ฟเวอร์ไม่ควรจะทำให้ประวัติข้อมูลหายไปจนบังคับให้ไคลเอนต์ทุกคนต้องเริ่มใหม่จากศูนย์

ให้ทำ index เหตุการณ์ของคุณด้วยจำนวนเต็มที่เพิ่มขึ้นอย่างต่อเนื่อง (monotonically increasing integer) หรือ ULID เมื่อมีการเชื่อมต่อใหม่เข้ามา ให้ query หาเหตุการณ์ที่ id > lastEventId แล้วส่งข้อมูลเหล่านั้นซ้ำตามลำดับ หากคุณมีข้อความค้างอยู่เป็นร้อยๆ ข้อความ ให้ใส่การหน่วงเวลาเล็กน้อยหรือส่งแบบเป็นชุด (batch) แต่ต้องส่งจากเก่าไปใหม่เพื่อให้ไคลเอนต์สามารถสร้างสถานะใหม่ตามลำดับเวลาได้

เตรียมรับมือกับข้อมูลซ้ำ

เครือข่ายนั้นไม่เสถียรเสมอไป เซิร์ฟเวอร์อาจส่ง event ออกไปแล้วสูญเสีย TCP acknowledgment ระหว่างทาง และส่งข้อมูลเดิมซ้ำอีกครั้งหลังจากหมดเวลา (timeout) ดังนั้นควรออกแบบระบบโดยคำนึงถึงการส่งข้อมูลแบบอย่างน้อยหนึ่งครั้ง (at-least-once delivery) ตั้งแต่เริ่มต้น

ในฝั่งไคลเอนต์ การกำจัดข้อมูลซ้ำ (deduplication) นั้นทำได้ง่ายและใช้ทรัพยากรน้อย ให้ใช้ Map โดยใช้ event ID เป็น key เมื่อมี event ใหม่เข้ามา ให้ตรวจสอบใน map หากพบ ID นั้นอยู่แล้ว ให้ทิ้งข้อมูลที่ซ้ำนั้นไปเงียบๆ เนื่องจากเซิร์ฟเวอร์ของคุณกำหนด ID แบบ deterministic ทำให้ข้อมูลที่ซ้ำกันไม่ก่อให้เกิดอันตราย ตัว map ไม่จำเป็นต้องขยายขนาดไปเรื่อยๆ เมื่อคุณยืนยันว่า event นั้นถูกประมวลผลอย่างปลอดภัยแล้ว ให้ลบ ID เก่าๆ ออก การใช้หน้าต่างแบบเลื่อน (sliding window) ขนาดไม่กี่ร้อยรายการมักจะเพียงพอสำหรับไคลเอนต์บนเบราว์เซอร์

เมื่อเคอร์เซอร์หมดอายุ

ในที่สุด ไคลเอนต์อาจจะกลับมาเชื่อมต่อใหม่หลังจากผ่านไปหลายชั่วโมงหรือหลายวัน หาก buffer ประวัติของคุณครอบคลุมเพียงแค่หนึ่งพันเหตุการณ์ล่าสุด แต่ไคลเอนต์ตามหลังอยู่ถึงสองพันเหตุการณ์ การส่งข้อมูลย้อนหลังเพื่อเติมเต็มช่องว่าง (gaps) ก็จะเป็นไปไม่ได้

อย่าส่งสตรีมประวัติเพียงบางส่วน เพราะจะทำให้ไคลเอนต์อยู่ในสถานะที่ไม่สอดคล้องกัน (inconsistent state) แต่ควรตรวจจับว่าเคอร์เซอร์หมดอายุแล้ว และส่ง snapshot ฉบับเต็มเป็น event ถัดไป โดย snapshot ควรมาพร้อมกับเคอร์เซอร์ใหม่ที่จะช่วยยึดไคลเอนต์ไว้กับสถานะปัจจุบัน หลังจากนั้นจึงค่อยส่งข้อมูลส่วนต่าง (deltas) แบบสดๆ ตามปกติ ควรระบุขอบเขตนี้ในโปรโตคอลของคุณให้ชัดเจน เพื่อให้โค้ดฝั่งไคลเอนต์รู้ว่าเมื่อใดควรจะรีเซ็ตโมเดลในเครื่องแทนที่จะเป็นการเพิ่มข้อมูลต่อท้าย

ปกป้องสตรีม

SSE endpoint ที่เปิดสาธารณะเป็นเป้าหมายที่น่าดึงดูด ใครก็ตามสามารถถือการเชื่อมต่อค้างไว้ได้ และการส่ง replay requests อาจเป็นการเพิ่มภาระการอ่านข้อมูล (read load) ในหน่วยจัดเก็บข้อมูลของคุณ

ควบคุมการเข้าถึง endpoint ด้วยการตรวจสอบสิทธิ์ที่เหมาะสม เนื่องจาก EventSource ของเบราว์เซอร์ไม่รองรับ custom headers ให้ส่ง token ผ่าน query string หรือใช้ cookies ที่มีนโยบาย SameSite แบบเข้มงวด (strict) แทน และควรตรวจสอบความถูกต้องของ token ก่อนที่จะจัดสรรทรัพยากรของ stream

กำหนดขีดจำกัดของประวัติ (history limits) และโควตาต่อผู้ใช้ (per-user quotas) จำกัดจำนวน event ที่จัดเก็บต่อหนึ่ง task และจำกัดจำนวนการเชื่อมต่อพร้อมกัน (concurrent connections) ต่อหนึ่ง client นอกจากนี้ควรบันทึก log การตัดการเชื่อมต่อ (disconnects) และการเล่นซ้ำ (replays) เพื่อให้คุณสามารถตรวจพบ client ที่ผิดปกติซึ่งพยายามยิง request ถล่ม cursor endpoint ของคุณได้

รูปแบบนี้สามารถนำไปประยุกต์ใช้ได้ในวงกว้าง

แนวทางนี้ไม่ได้จำกัดอยู่แค่ใน HTTP เท่านั้น กฎเกณฑ์เดียวกันนี้ยังสามารถนำไปใช้ได้เมื่อคุณเปลี่ยนไปใช้ WebSockets, message queues หรือ agent-to-agent interfaces แม้ว่าวิธีการรับส่งข้อมูล (transport) จะเปลี่ยนไป—เช่น การใช้ binary frames หรือ topic subscriptions—แต่ปัญหาพื้นฐานยังคงเหมือนเดิม คุณยังคงต้องการ cursor, durable log, หลักการทำงานแบบ at-least-once semantics, การทำ client deduplication และการสำรองข้อมูลด้วย full snapshots เมื่อ cursor ไม่เป็นปัจจุบันอีกต่อไป เมื่อคุณแก้ปัญหาเรื่อง state convergence ได้แล้ว คุณก็สามารถส่งข้อมูลผ่าน TCP, WebSocket หรือ broker อย่าง RabbitMQ ได้โดยไม่ต้องออกแบบตรรกะหลัก (core logic) ใหม่

เน้นความเรียบง่าย

Server-Sent Events ทำงานได้ดีเพราะทำงานบน HTTP ปกติ Proxy สามารถเข้าใจได้ Load balancer สามารถทำ health-check ได้ และการ debug ก็ง่ายเหมือนการใช้ curl แต่ความเรียบง่ายนั้นจะหายไปหากคุณละเลยกรณีขอบเขต (edge cases) จงสร้าง cursor, เตรียมรับมือกับการ replay, ทำ deduplication ที่ฝั่ง client และทำ snapshot เมื่อประวัติหมดลง หากทำเช่นนี้ งาน AI ที่ต้องใช้เวลานานของคุณจะรายงานความคืบหน้าได้อย่างแม่นยำ แม้จะผ่าน Wi-Fi ที่ไม่เสถียร, การรีสตาร์ทเซิร์ฟเวอร์ หรือแม้แต่กรณีที่เบราว์เซอร์เข้าสู่โหมด sleep ในช่วงข้ามคืน

ที่มา: Build a Reconnecting SSE Task Stream with Node.js

เข้าร่วมการสนทนา: GyaanSetu AI Community