Honker - Phần mở rộng thêm ngữ nghĩa NOTIFY/LISTEN của Postgres vào SQLite

 (github.com/russellromney)
1 điểm bởi GN⁺ 11 giờ trước | 1 bình luận | Chia sẻ qua WhatsApp
  • Thông qua phần mở rộng SQLite và các binding cho nhiều ngôn ngữ, có thể xử lý cùng lúc durable pub/sub, hàng đợi tác vụ và luồng sự kiện ngay trong cùng một tệp .db mà không cần client polling hay daemon/broker riêng
  • notify(), stream(), queue() đều được ghi trong transaction của chính bên gọi, nên sẽ commit cùng hoặc rollback cùng với ghi chép nghiệp vụ, giúp giảm vấn đề dual-write
  • Cơ chế đánh thức giữa các tiến trình hoạt động bằng cách kiểm tra PRAGMA **data_version** mỗi 1ms, được tinh chỉnh để hướng tới độ trễ mức mili giây một chữ số và chi phí truy vấn rất nhỏ
  • Hàng đợi tác vụ bao gồm retry, priority, delayed execution, dead-letter, scheduler, named lock, rate limiting; còn stream hỗ trợ at-least-once delivery với offset được lưu theo từng consumer
  • Trong môi trường dùng SQLite làm kho lưu trữ chính, đây là cấu hình gộp ứng dụng và xử lý bất đồng bộ vào một tệp cơ sở dữ liệu, giúp giảm độ phức tạp vận hành; API hiện vẫn ở trạng thái Experimental

Tổng quan

  • Bằng phần mở rộng SQLite và các binding cho nhiều ngôn ngữ, Honker bổ sung cơ chế NOTIFY/LISTEN kiểu Postgres vào SQLite, đồng thời cho phép xử lý durable pub/sub, hàng đợi tác vụ và luồng sự kiện trong cùng một tệp .db mà không cần client polling hay daemon/broker riêng
  • Dựa trên layout on-disk được định nghĩa một lần bằng Rust, các binding Python, Node, Bun, Ruby, Go, Elixir, C++ đều được thiết kế như các lớp bao mỏng quanh cùng một loadable extension
  • Thay thế polling ở tầng ứng dụng bằng cách đọc cơ sở dữ liệu mỗi 1ms; chi phí truy vấn PRAGMA data_version ở mức micro giây một chữ số và việc truyền thông báo giữa các tiến trình được tinh chỉnh ở mức mili giây một chữ số
  • Khi dùng SQLite làm kho lưu trữ chính, có thể commit hoặc rollback ghi chép nghiệp vụ và việc đưa tác vụ vào hàng đợi trong cùng một transaction, giúp giảm nhu cầu vận hành datastore riêng và vấn đề dual-write
  • API hiện vẫn ở trạng thái Experimental và có thể thay đổi
  • Tác giả cũng nói rõ rằng nếu đã vận hành Postgres, thì dùng pg_notify, pg-boss, Oban sẽ phù hợp hơn

Tính năng chính

  • Cung cấp đồng thời notify/listen giữa các tiến trình, hàng đợi tác vụ có retry, priority, delayed execution, dead-letter table, và durable stream với offset theo từng consumer trong cùng một tệp .db
  • Mọi thao tác gửi đều có thể kết hợp nguyên tử với ghi chép nghiệp vụ, nên sẽ commit cùng hoặc rollback cùng nhau
  • Thời gian phản hồi liên tiến trình ở mức mili giây một chữ số; đồng thời cũng bao gồm handler timeout, retry dựa trên exponential backoff, delayed jobs, task expiration, named lock, rate limiting
  • Cũng hỗ trợ scheduler dựa trên leader election và periodic task kiểu crontab, cùng khả năng lưu task result theo cơ chế opt-in
    • enqueue trả về id, worker sẽ lưu giá trị trả về, và bên gọi có thể chờ kết quả bằng queue.wait_result(id)
  • Được cung cấp dưới dạng SQLite loadable extension nên bất kỳ client SQLite nào cũng có thể đọc cùng các bảng này
  • Hoạt động cả bên trong kết nối SQLite do ORM sở hữu; hướng dẫn ORM đề cập tích hợp với SQLAlchemy, SQLModel, Django, Drizzle, Kysely, sqlx, GORM, ActiveRecord, Ecto
  • Đồng thời cũng nêu rõ những phạm vi cố ý không hỗ trợ
    • Không hỗ trợ task pipeline, chain, group, chord
    • Không hỗ trợ multi-writer replication
    • Không hỗ trợ workflow orchestration dựa trên DAG

Bắt đầu nhanh

  • Hàng đợi Python

    • Mở cơ sở dữ liệu bằng honker.open("app.db") và lấy hàng đợi như db.queue("emails") để đưa công việc vào hàng và tiêu thụ chúng
    • Trong khối with db.transaction() as tx:, nếu thực hiện INSERT đơn hàng và emails.enqueue(..., tx=tx) cùng nhau thì việc ghi đơn hàng và đưa tác vụ gửi mail vào hàng sẽ được gói trong cùng một transaction
    • Worker lấy công việc từng cái một dưới dạng async for job in emails.claim("worker-1"):; khi thành công thì gọi job.ack(), khi thất bại thì xử lý bằng job.retry(delay_s=60, error=str(e))
    • claim() là một async iterator và nội bộ sẽ gọi claim_batch(worker_id, 1) ở mỗi vòng lặp
    • Nó sẽ thức dậy với bất kỳ commit nào vào cơ sở dữ liệu, và chỉ quay về paranoia poll 5 giây khi commit watcher không thể hoạt động
    • Tác vụ theo lô được tách riêng để dùng trực tiếp claim_batch(worker_id, n) và queue.ack_batch(ids, worker_id); visibility mặc định là 300 giây

  • Tác vụ Python

    • Dùng decorator @emails.task(retries=3, timeout_s=30) thì lời gọi hàm sẽ được chuyển trực tiếp thành việc đưa vào hàng đợi và trả về TaskResult
    • Bên gọi có thể dùng như send_email("alice@example.com", "Hi") và chờ kết quả thực thi từ worker bằng r.get(timeout=10)
    • Worker có thể chạy dưới dạng tiến trình riêng hoặc in-process như python -m honker worker myapp.tasks:db --queue=emails --concurrency=4
    • Tên tự động là {module}.{qualname}; trong môi trường production, nên dùng tên tường minh như @emails.task(name="...") để tránh việc đổi tên làm các pending job bị mồ côi
    • Tác vụ định kỳ dùng dạng @emails.periodic_task(crontab("0 3 * * *"))
    • Ví dụ chi tiết có trong packages/honker/examples/tasks.py

  • Stream Python

    • db.stream("user-events") cung cấp pub/sub bền vững, và có thể thực hiện UPDATE nghiệp vụ cùng với stream.publish(..., tx=tx) trong cùng một transaction
    • Nếu đăng ký bằng async for event in stream.subscribe(consumer="dashboard"): thì nó sẽ phát lại các hàng sau offset đã lưu, rồi sau đó chuyển sang truyền thời gian thực dựa trên commit
    • Offset của mỗi named consumer được lưu trong bảng _honker_stream_consumers
    • Việc tự động lưu offset mặc định chỉ diễn ra mỗi 1000 sự kiện hoặc mỗi 1 giây một lần, để không đập quá nhiều vào single-writer slot ngay cả khi throughput cao
    • Có thể điều chỉnh bằng save_every_n= và save_every_s=; nếu đặt cả hai về 0 thì sẽ tắt tự động lưu và có thể tự gọi stream.save_offset(consumer, offset, tx=tx)
    • Nếu xảy ra crash, các sự kiện in-flight sau offset đã được flush lần cuối sẽ được gửi lại, theo mô hình at-least-once

  • Python notify

    • Có thể đăng ký pub/sub tạm thời bằng async for n in db.listen("orders"): và gửi thông báo bằng tx.notify("orders", {"id": 42}) bên trong transaction
    • Listener hiện sẽ bám từ vị trí MAX(id) hiện tại nên không phát lại lịch sử cũ
    • Nếu cần durable replay thì phải dùng db.stream()
    • Bảng notifications không được dọn tự động, vì vậy cần gọi db.prune_notifications(older_than_s=…, max_keep=…) trong tác vụ được lên lịch
    • Payload của task phải hợp lệ dưới dạng JSON, và Python writer cùng Node reader có thể dùng chung một channel

  • Node.js

    • Trong binding Node cũng dùng cùng mẫu tính năng với open('app.db'), db.transaction(), tx.notify(...), db.listen('orders')
    • Ghi nghiệp vụ và notify được gói trong cùng một commit, còn listen sẽ thức dậy với bất kỳ commit nào vào cơ sở dữ liệu rồi lọc theo channel

  • SQLite extension

    • Sau .load ./libhonker_ext, khởi tạo bằng SELECT honker_bootstrap(); và chỉ với các hàm SQL cũng có thể dùng hàng đợi, khóa, rate limit, scheduler, stream, và lưu kết quả
    • Cung cấp các hàm như honker_claim_batch, honker_ack_batch, honker_sweep_expired, honker_lock_acquire, honker_rate_limit_try, honker_scheduler_tick, honker_stream_publish, honker_stream_read_since, honker_result_save
    • Python binding và extension cùng chia sẻ _honker_live, _honker_dead, _honker_notifications, nên worker Python có thể lấy các tác vụ mà ngôn ngữ khác đã đưa vào bằng extension
    • Tính tương thích schema được cố định trong tests/test_extension_interop.py

Thiết kế

  • Kho lưu trữ này đồng thời bao gồm SQLite loadable extension honker cùng các binding cho Python, Node, Rust, Go, Ruby, Bun, Elixir
  • Nó nhắm đến các ứng dụng dùng SQLite làm kho lưu trữ chính, và tập trung vào việc chuyển package logic vào SQLite extension để có thể dùng theo cách tương tự trên nhiều ngôn ngữ và framework
  • Có ba primitive cốt lõi
    • notify() là pub/sub tạm thời
    • stream() là pub/sub bền vững có offset theo từng consumer
    • queue() là hàng đợi công việc at-least-once
  • Cả ba primitive này đều được ghi bằng INSERT trong transaction của bên gọi, nên việc chuyển tác vụ và ghi dữ liệu nghiệp vụ sẽ cùng commit hoặc cùng rollback
  • Mục tiêu là hiện thực hành vi tương tự NOTIFY/LISTEN mà không cần polling ở cấp ứng dụng, để đạt thời gian phản hồi nhanh
  • Nếu dùng nguyên file SQLite hiện có, mọi commit vào cơ sở dữ liệu sẽ đánh thức worker, và phần lớn trigger có thể chỉ đọc message hoặc queue rồi kết thúc với kết quả rỗng mà không xử lý gì thực sự
  • Hiện tượng overtriggering này là một tradeoff có chủ đích, được chọn để đổi lấy hành vi gần với push và thời gian phản hồi nhanh

Giá trị mặc định khuyến nghị cho WAL

  • Các language binding mặc định sử dụng journal_mode = WAL, cung cấp mô hình nhiều reader đồng thời và một writer duy nhất, fsync batching hiệu quả, cùng thiết lập wal_autocheckpoint = 10000
  • Các mode khác như DELETE, TRUNCATE, MEMORY cũng hoạt động, và việc phát hiện commit được thực hiện dựa trên PRAGMA data_version tăng dần trong mọi journal mode
  • Điều mất đi ở mode không dùng WAL chỉ là đặc tính ghi trong khi đang đọc đồng thời; tính đúng đắn và bản thân việc đánh thức giữa các tiến trình không phụ thuộc vào WAL
  • Toàn bộ hệ thống gồm một tệp .db, và khi bật WAL thì có thể xuất hiện thêm các sidecar .db-wal, .db-shm
  • claim được xử lý bằng một lệnh UPDATE … RETURNING duy nhất thông qua partial index, còn ack bằng một lệnh DELETE duy nhất
  • Ở bất kỳ journal mode nào cũng chỉ có một writer tại một thời điểm; lợi ích về reader đồng thời là do WAL cung cấp
  • PRAGMA data_version tăng sau mỗi commit và checkpoint, nên xử lý đúng cả các tình huống như WAL truncation, tạo và xóa journal file, hay tái sử dụng cùng kích thước
  • SQLite không có wire protocol nên không thể push từ server; consumer phải tự bắt đầu đọc
    • tín hiệu wake là việc tăng counter
    • sau đó việc truy vấn thực tế là SELECT
  • Vì transaction có chi phí thấp, jobs, events và notifications được ghi trong khối with db.transaction() đang mở của caller theo kiểu outbox pattern
  • Thay vì dùng stat(2) để xem kích thước·mtime của tệp WAL hay kernel watcher như FSEvents·inotify·kqueue, hệ thống sử dụng PRAGMA data_version
    • data_version là monotonic counter mà SQLite tăng lên khi có commit từ bất kỳ connection nào
    • xử lý đúng WAL truncation, clock skew và các transaction đã rollback
    • kernel watcher trên macOS có thể bỏ sót ghi từ cùng tiến trình, và stat(2) dựa trên (size, mtime) có thể bỏ lỡ commit khi WAL bị truncate rồi lại tăng lên đúng cùng kích thước
    • hoạt động giống nhau trên Linux, macOS và Windows, và chi phí CPU ở độ phân giải mức 1ms là rất nhỏ
    • chi phí mỗi truy vấn khoảng 3.5µs, tức tổng khoảng 3.5ms/giây ở mức 1kHz
  • Mô hình khóa của SQLite giả định single machine, single writer; nếu hai server cùng ghi vào một .db trên NFS thì sẽ bị hỏng
    • trong trường hợp này cần shard theo mức tệp hoặc chuyển sang Postgres

Kiến trúc

  • Đường wake

    • Mỗi Database có một PRAGMA poll thread để truy vấn data_version mỗi 1ms
    • Khi counter thay đổi, hệ thống fan-out tick tới bounded channel của từng subscriber
    • Mỗi subscriber chạy SELECT … WHERE id > last_seen tận dụng partial index, trả về các hàng mới rồi chờ tiếp
    • Dù có 100 subscriber thì cũng chỉ cần 1 poll thread
    • listener ở trạng thái idle không chạy bất kỳ truy vấn SQL nào
    • chi phí idle chỉ là một truy vấn PRAGMA data_version mỗi 1ms cho mỗi database, và số lượng listener tăng gần như miễn phí nhờ cấu trúc dùng SQLite counter read
    • SharedWalWatcher của honker-core sở hữu poll thread và fan-out qua các kênh bounded SyncSender<()> theo từng subscriber id
    • Mỗi lần gọi db.wal_events() sẽ đăng ký subscriber, và handle được trả về sẽ tự động hủy đăng ký khi Drop
    • Khi listener bị drop, rx.recv() -> Err sẽ xảy ra ở bridge thread, sau đó nó dọn dẹp rồi thoát

  • Lược đồ queue

    • _honker_live chứa các hàng ở trạng thái pending và processing
    • partial index có dạng (queue, priority DESC, run_at, id) WHERE state IN ('pending','processing')
    • claim được thực hiện bằng một lệnh UPDATE … RETURNING duy nhất thông qua index này
    • ack là một lệnh DELETE duy nhất
    • Các hàng vượt quá giới hạn retry được chuyển sang _honker_dead và không bị quét lại trong đường claim
    • Nhờ partial index trên state, đường nóng của claim bị giới hạn bởi kích thước working set chứ không phải toàn bộ kích thước history
    • Dù có 100k dead row thì tốc độ claim vẫn giữ nguyên như queue không có dead row

  • Iterator claim

    • async for job in q.claim(id) lặp lại việc gọi claim_batch(id, 1) và trả ra từng tác vụ một
    • Job.ack() là một lệnh DELETE duy nhất trong transaction riêng của nó; giá trị trả về là True nếu claim vẫn còn hiệu lực, hoặc False nếu visibility window đã qua và worker khác đã lấy lại
    • Hệ thống thức dậy với mọi database commit từ bất kỳ tiến trình nào, và paranoia poll 5 giây là fallback duy nhất
    • Với xử lý theo batch, phải dùng trực tiếp claim_batch(worker_id, n) và queue.ack_batch(ids, worker_id)
    • Thư viện không ẩn batch phía sau iterator, để xử lý rõ ràng hơn chi phí transaction và hành vi visibility at-most-once

  • Gắn với transaction

    • notify() là SQL scalar function được đăng ký trên writer connection
    • Nó thực hiện INSERT vào _honker_notifications trong transaction đang mở của caller
    • queue.enqueue(…, tx=tx) và stream.publish(…, tx=tx) cũng hoạt động theo cùng cách
    • Nếu xảy ra rollback thì job, event và notification cũng biến mất theo
    • Đây là transactional outbox pattern được cung cấp sẵn, cho phép xử lý cùng lúc business write và enqueue side effect mà không cần cài thêm library riêng
    • Không có dispatch table hay dispatcher process riêng; chính side effect row là hàng đã được commit, và bất kỳ tiến trình nào đang theo dõi database đều có thể nhặt nó lên trong khoảng 1ms

  • Over-triggering nhanh hơn polling

    • Mọi thay đổi của data_version đều đánh thức tất cả subscriber của Database đó, chứ không đánh thức chọn lọc theo channel đã commit
    • Chi phí của việc bị đánh thức nhầm chỉ là một lần SELECT có index ở mức micro giây
    • Ngược lại, bỏ sót đối tượng cần đánh thức sẽ dẫn đến lỗi correctness âm thầm
    • Việc lọc theo channel được xử lý ở đường SELECT, không phải ở giai đoạn trigger notification
    • SQLite cũng có thể xử lý hiệu quả mẫu thực hiện nhiều truy vấn nhỏ

  • Chính sách lưu giữ

    • Tác vụ trong queue được giữ lại cho đến khi được ack, và nếu vượt quá giới hạn retry thì được chuyển sang _honker_dead
    • Event trong stream được giữ lại, và mỗi named consumer tự theo dõi offset của mình
    • notify là fire-and-forget và không có dọn dẹp tự động
    • Chính sách lưu giữ được caller chọn cho từng primitive, và phải tự gọi db.prune_notifications(older_than_s=…, max_keep=…)
    • Cách làm này khiến chính sách retention hiển lộ trong mã của caller thay vì bị ẩn sau các giá trị mặc định của thư viện

Khôi phục sau sự cố

  • rollback sẽ xóa toàn bộ jobs, events, notifications cùng với các ghi nghiệp vụ theo đặc tính ACID của SQLite
  • vẫn an toàn ngay cả khi bị SIGKILL giữa giao dịch, và khi mở lại lần sau, atomic commit rollback của SQLite sẽ không để lại trạng thái cũ
    • việc dùng WAL hay rollback journal phụ thuộc vào journal mode
    • việc kiểm chứng được thực hiện trong tests/test_crash_recovery.py, bằng cách kết thúc subprocess trước COMMIT rồi xác nhận PRAGMA integrity_check == 'ok' và luồng notify mới
  • nếu worker chết giữa lúc xử lý, sau khi visibility_timeout_s trôi qua thì worker khác sẽ claim lại
    • giá trị mặc định là 300 giây
    • attempts sẽ tăng lên
    • nếu vượt quá mặc định max_attempts là 3 lần thì hàng sẽ được chuyển sang _honker_dead
  • listener đang offline trong lúc prune sẽ bỏ lỡ các event đã bị xóa; nếu cần durable replay thì nên dùng db.stream() để lưu offset theo từng consumer

Tích hợp web framework

  • không cung cấp plugin cho framework, thay vào đó chọn cách kết nối bằng vài dòng glue code vì API nhỏ gọn
  • với FastAPI, có ví dụ khởi chạy worker loop khi startup, và trong lúc xử lý request thì thực hiện business write cùng queue enqueue bên trong cùng một transaction
  • endpoint SSE có thể được dựng trong khoảng 30 dòng theo dạng async def stream(...): yield f"data: ...\n\n" trên db.listen(channel) hoặc db.stream(name).subscribe(...)
  • với Django và Flask, khuyến nghị cấu hình chạy worker như một tiến trình CLI riêng theo kiểu Celery hoặc RQ

Sử dụng ORM

  • trên kết nối ORM, load libhonker_ext rồi gọi các hàm SQL trong chính transaction của ORM thì enqueue sẽ được commit một cách nguyên tử cùng với business write
  • trong ví dụ SQLAlchemy, extension được load ở sự kiện connect và chạy SELECT honker_bootstrap(), sau đó bên trong transaction s.begin() sẽ gọi cùng lúc model INSERT và SELECT honker_enqueue(...)
  • worker chạy trong một tiến trình riêng dùng honker.open("app.db"), và commit watcher sẽ thức dậy với mọi commit từ bất kỳ kết nối nào tới cùng file đó
  • hướng dẫn Using with an ORM bao gồm tích hợp Django, SQLModel, Drizzle, Kysely, sqlx, GORM, ActiveRecord, Ecto, mẫu wrapper TypedQueue[T] cho SQLModel/Pydantic, cùng các caveat liên quan đến Prisma

Hiệu năng

  • cho biết có thể xử lý hàng nghìn message mỗi giây trên laptop hiện đại
  • độ trễ wake liên tiến trình bị giới hạn bởi 1ms poll cadence, và theo M-series thì median vào khoảng 1~2ms
  • có thể đo trên phần cứng thực bằng bench/wake_latency_bench.py và bench/real_bench.py

Cấu hình phát triển

  • Bố cục kho lưu trữ

    • honker-core/: Rust rlib được mọi binding dùng chung, được đưa trực tiếp vào cây mã và cũng được phát hành trên crates.io
    • honker-extension/: cdylib cho SQLite loadable extension, được đưa trực tiếp vào cây mã và cũng được phát hành trên crates.io
    • packages/honker/: gói Python bao gồm PyO3 cdylib cùng Queue, Stream, Outbox, Scheduler
    • packages/honker-node/: binding Node.js, là git submodule
    • packages/honker-rs/: wrapper tiện dụng cho Rust, là git submodule
    • packages/honker-go/: binding Go, là git submodule
    • packages/honker-ruby/: binding Ruby, là git submodule
    • packages/honker-bun/: binding Bun, là git submodule
    • packages/honker-ex/: binding Elixir, là git submodule
    • packages/honker-cpp/: binding C++, là git submodule
    • tests/: thư mục integration test cross-package
    • bench/: thư mục benchmark
    • site/: trang honker.dev, dựa trên Astro và là git submodule
    • mỗi kho binding được phát hành riêng lên PyPI, npm, crates.io, Hex, RubyGems..., còn phần nền tảng dùng chung là honker-core và honker-extension thì được chứa trực tiếp trong kho này
    • khi clone cần dùng git clone --recursive hoặc git submodule update --init --recursive

Kiểm thử và độ bao phủ

  • make test mặc định chạy test Rust, Python, Node và mất khoảng 10 giây theo đường nhanh
  • make test-python-slow bao gồm soak và real-time cron test, mất khoảng 2 phút
  • make test-all chạy toàn bộ test, gồm cả các mark chậm
  • make build thực hiện PyO3 maturin develop và build loadable extension
  • benchmark có thể chạy bằng python bench/wake_latency_bench.py --samples 500, python bench/real_bench.py --workers 4 --enqueuers 2 --seconds 15, python bench/ext_bench.py
  • cài công cụ coverage dùng make install-coverage-deps, sẽ cài coverage.py và cargo-llvm-cov
  • make coverage tạo hai báo cáo HTML trong coverage/, còn make coverage-python tạo báo cáo cho nhánh Python, make coverage-rust tạo báo cáo dựa trên Rust unit test của honker-core
  • ghi chú rằng coverage Python cho packages/honker/ vào khoảng 92%
  • coverage Rust chỉ phản ánh cargo test; nhiều nhánh trong honker_ops.rs chỉ được chạy bởi bộ test Python nên không xuất hiện trong báo cáo Rust
  • việc kết hợp cross-language coverage thông qua hợp nhất dữ liệu profile LLVM vượt qua ranh giới PyO3 là khó và hiện vẫn đang bị trì hoãn

Giấy phép

  • sử dụng giấy phép Apache 2.0
  • xem chi tiết tại LICENSE

Bài viết liên quan

1 bình luận

 
GN⁺ 11 giờ trước
Ý kiến trên Hacker News

  • Tôi đã tự làm cái này. Honker thêm NOTIFY/LISTEN liên tiến trình vào SQLite, để truyền sự kiện kiểu push với độ trễ chỉ vài ms bằng chính file SQLite sẵn có, không cần daemon hay broker
    SQLite không có server như Postgres, nên điểm mấu chốt là chuyển nguồn polling sang stat(2) nhẹ trên file WAL thay vì query theo chu kỳ. SQLite vẫn hiệu quả ngay cả khi bắn nhiều query nhỏ (https://www.sqlite.org/np1queryprob.html), nên khó gọi đây là một nâng cấp khổng lồ, nhưng việc chỉ cần theo dõi WAL và gọi hàm SQLite khiến nó không phụ thuộc ngôn ngữ là điều khá thú vị
    Tôi cũng thêm ephemeral pub/sub, durable work queue có retry và dead-letter, cùng event stream có offset theo từng consumer. Cả ba đều chỉ là các row trong file .db của ứng dụng hiện có, nên có thể commit nguyên tử cùng với ghi nghiệp vụ, và nếu rollback thì cả hai cùng biến mất
    Ban đầu nó tên là litenotify/joblite, nhưng sau khi mua honker.dev cho vui thì thấy Oban, pg-boss, Huey, RabbitMQ, Celery, Sidekiq đều có tên khá ngộ nên tôi cứ chốt tên này. Hy vọng nó hữu ích hoặc ít nhất là buồn cười, và cảnh báo đây là phần mềm alpha vẫn hoàn toàn đúng

    • Có vẻ cái này chủ yếu phù hợp với các ngôn ngữ mà chỉ dễ xử lý đồng thời dựa trên tiến trình
      Với Java/Go/Clojure/C# thì SQLite vốn dĩ đã là single writer, nên có vẻ đơn giản và gọn hơn nếu ứng dụng tự quản lý writer đó rồi dùng concurrent queue ở cấp ngôn ngữ để biết có ghi nào xảy ra và chỉ đánh thức các thread liên quan
      Dù vậy, cách tận dụng WAL sáng tạo như thế này vẫn rất thú vị, và với các ngôn ngữ như Python/JS/TS/Ruby, nơi đồng thời dựa trên tiến trình là phổ biến, nó có vẻ khá hợp làm cơ chế notify
    • Lần này tôi mới biết stat() mỗi 1ms hóa ra rẻ hơn nhiều so với tưởng tượng
      Trên phần cứng của tôi, mỗi lần gọi còn chưa đến 1μs, nên kiểu polling này dùng chưa đến 0,1% CPU
    • Có thể tôi đã bỏ sót gì đó, nhưng liệu PRAGMA data_version có tốt hơn stat(2) không nhỉ
      https://sqlite.org/pragma.html#pragma_data_version
      Nếu là C API thì còn có SQLITE_FCNTL_DATA_VERSION trực tiếp hơn nữa
      https://sqlite.org/c3ref/c_fcntl_begin_atomic_write.html#sqlitefcntldataversion
    • Khá hay. Tôi cũng từng làm dở một thứ tương tự
      Tôi tò mò liệu cái này có thể dùng như Kafka hạng nhẹ cho luồng thông điệp bền vững hay không. Cũng muốn biết liệu có hỗ trợ ngữ nghĩa kiểu replay toàn bộ thông điệp quá khứ + thời gian thực từ một timestamp nào đó cho một topic cụ thể không
      Có lẽ vẫn có thể giả lập bằng polling như pub/sub, nhưng như bạn nói thì chắc không phải tối ưu
    • Có lẽ sẽ còn tốt hơn nếu lưu cả trạng thái subscriber
      Nếu lưu vị trí đọc, tên queue, bộ lọc v.v. thì thay vì mỗi khi stat(2) đổi là đánh thức mọi subscription thread để mỗi thread tự chạy một SELECT với N=1, polling thread có thể Events INNER JOIN Subscribers và chỉ đánh thức những subscriber thật sự khớp

  • Cảm ơn phản hồi. Tôi đã mở PR phản ánh các đề xuất đó
    https://github.com/russellromney/honker/pulls/1
    Giờ nó đã chuyển sang cấu trúc polling 3 tầng: PRAGMA data_version mỗi 1ms, stat mỗi 100ms, và xử lý reconnect khi có lỗi

    1. Mỗi 1ms dùng PRAGMA data_version để thay thế việc phát hiện thay đổi size/mtime dựa trên stat trước đây. Đây là commit counter của chính SQLite nên đơn điệu tăng, không bị ảnh hưởng bởi clock skew, và xử lý đúng cả WAL truncation lẫn rollback. Nó là một query nonblocking khoảng 3µs, và tôi đổi không phải vì hiệu năng mà vì độ chính xác. Thậm chí nó còn chậm hơn một chút. Rủi ro truncation hóa ra cũng thực tế hơn tôi tưởng
      Khi test thì C API SQLITE_FCNTL_DATA_VERSION không hoạt động giữa các connection. Vì vậy hiện giờ tôi vẫn phải chấp nhận chi phí đi qua VFS layer, và đang chủ động chấp nhận tradeoff đó
    2. Nếu query data_version thất bại thì sẽ thử reconnect, giả định các trường hợp như lỗi đĩa tạm thời, NFS hiccup, connection corruption, và để phòng ngừa thì cũng đánh thức subscriber
    3. Mỗi 100ms dùng stat để so sánh (dev, ino) với giá trị lúc startup nhằm phát hiện thay thế file. Đây là các trường hợp như atomic rename, litestream restore, volume remount; data_version đi theo fd đang mở nên nếu file đã bị thay thì nó vẫn tiếp tục nhìn inode cũ và không bắt được việc này
      Nhờ vậy Honker tốt hơn hẳn và tôi cũng học được rất nhiều

  • Nhân tiện quảng bá nhẹ, trong PostgreSQL 19 sắp tới, LISTEN/NOTIFY đã được tối ưu để scale tốt hơn nhiều cho selective signaling
    Đây là bản vá nhắm vào trường hợp có nhiều backend cùng listen các channel khác nhau
    https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=282b1cde9

    • Quảng bá rất ổn, mà còn cực kỳ đúng chủ đề nữa

  • Tôi tự hỏi sao không theo dõi thay đổi WAL bằng inotify hoặc wrapper đa nền tảng để khỏi polling

    • Thế là hỏng đa nền tảng. Đặc biệt trên Mac có trường hợp nó âm thầm nuốt mất, nên khó mà tin cậy
      stat thì đơn giản là chạy ở khắp mọi nơi

  • Điều hấp dẫn hơn so với IPC riêng là nó được commit nguyên tử với dữ liệu nghiệp vụ
    Với cơ chế chuyển thông điệp bên ngoài lúc nào cũng có vấn đề kiểu “đã gửi thông báo nhưng transaction lại rollback”, và mọi thứ rất nhanh trở nên rối rắm
    Tôi có một thắc mắc về WAL checkpoint. Khi SQLite truncate WAL về lại 0 thì polling bằng stat() có xử lý đúng không. Tôi có cảm giác sẽ có một khoảng trống làm mất sự kiện

    • Tôi nghĩ tính nguyên tử gần như là tất cả
      Trước đây tôi từng rất khổ với tổ hợp Postgres+SQS vì trigger gửi enqueue trước khi commit trên connection khác trở nên nhìn thấy được. Phải thêm retry logic, thêm polling phía worker, cuối cùng lại phải đưa enqueue vào trong transaction, mà làm vậy thì rốt cuộc chỉ là dựng lại điều Honker đang làm nhưng với nhiều moving part hơn
      Các lỗi kiểu “notification đã gửi nhưng row vẫn chưa commit” thường im lặng và phụ thuộc timing, nên truy vết thực sự rất đau đầu
    • File WAL vẫn còn đó và chỉ bị truncate nên bản thân việc đó vẫn được bắt như một update
      Tuy nhiên phần này tôi vẫn chưa có test nên cần kiểm tra thêm. Đây là một điểm hay, tôi sẽ xem kỹ

  • Cảm ơn
    Các ứng dụng nhỏ dựa trên SQLite đã tăng lên rất nhiều, và phần lớn đều cần queue và scheduler
    Tôi đã tự vận hành vài thứ, nhưng lúc nào cũng thấy thiếu sự tao nhã của các giải pháp kiểu Postgres
    Tôi định sẽ thử cái này ngay

    • Cụm “sự sinh sôi nhỏ” mô tả quá đúng cái quần thể do thói quen làm side project của tôi tạo ra
      Nếu gặp vấn đề thì rất mong bạn mở PR hoặc issue trong repo

  • Ở đây tôi rất muốn dùng kqueue/FSEvents, nhưng tôi nhớ là Darwin sẽ làm rơi thông báo từ cùng một tiến trình
    Nếu publisher và listener ở cùng tiến trình thì có trường hợp listener hoàn toàn không được đánh thức, nên việc truy vết khá bẩn. Polling bằng stat nhìn thì xấu nhưng rốt cuộc lại là thứ thực sự chạy được ở mọi nơi
    Tôi cũng tò mò liệu khi WAL checkpoint làm file nhỏ lại thì có phát sinh wakeup không, hay poller lọc luôn trường hợp giảm kích thước

    • Bình luận này hoàn toàn sai
      Sự kiện VNODE của kqueue sẽ được gửi miễn là tiến trình đó có quyền truy cập file, không có bộ lọc nào loại vì cùng tiến trình cả
    • Cái này thực sự cần test
      Tôi sẽ kiểm tra rồi báo lại

  • Rất hay. Tôi tò mò khi có tải thì nút thắt cổ chai chủ yếu là throughput ghi của SQLite, hay là lớp thông báo WAL

    • Nút thắt là phía ghi và luồng claim/ack
      Nó cũng thay đổi khá nhiều tùy journal mode và synchronous mode
      Phần notification thì cực kỳ rẻ, dù là cách cũ dùng stat(2) hay cách mới dựa trên PRAGMA. Ở bình luận khác cũng đã nói stat(2) cỡ khoảng 1µs

  • Dự án hay đấy. Tôi cũng đang làm một thứ đẩy SQLite đi xa hơn rất nhiều so với cách nó thường được dùng
    Thật đáng khích lệ khi thấy ngày càng nhiều người khám phá xem SQLite thực sự có thể làm được đến đâu

  • Tôi tò mò liệu có thể tích hợp cả khi dùng SQLAlchemy hay không
    Nhìn hiện tại thì có vẻ nó muốn tự tạo DB connection riêng