Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Nixevol
|
1d6be59652 |
@@ -0,0 +1,196 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/cookiejar"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.asio.asia/nixevol/NixMsg/internal/config"
|
||||
)
|
||||
|
||||
// TestUplinkDisableFatalAndRevoked 验证停用在线端收到 fatal，已推送投递收到 revoked。
|
||||
func TestUplinkDisableFatalAndRevoked(t *testing.T) {
|
||||
dataDir := t.TempDir()
|
||||
cfgPath := writeTestConfig(t, dataDir)
|
||||
initAdminForTest(t, dataDir)
|
||||
enableRegistration(t, dataDir, "uplink-code")
|
||||
|
||||
cfg, err := config.Load(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if vErr := cfg.Validate(); vErr != nil {
|
||||
t.Fatal(vErr)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- runServe(ctx, cfg) }()
|
||||
defer func() {
|
||||
cancel()
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if err != nil {
|
||||
t.Errorf("serve exit: %v", err)
|
||||
}
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Error("serve did not stop")
|
||||
}
|
||||
}()
|
||||
|
||||
addr := waitListenAddr(t, dataDir, 15*time.Second)
|
||||
base := "http://" + addr
|
||||
|
||||
registerEP(t, base, "alice", "password12", "Alice")
|
||||
registerEP(t, base, "bob", "password12", "Bob")
|
||||
|
||||
alice := mqttSessionLogin(t, base, "alice", "password12")
|
||||
defer alice.Close()
|
||||
bob := mqttSessionLogin(t, base, "bob", "password12")
|
||||
defer bob.Close()
|
||||
|
||||
delay0 := int64(0)
|
||||
sendResp := alice.Request(t, map[string]any{
|
||||
"v": 1, "type": "send", "rid": "s1", "id": "dm-fatal-1",
|
||||
"to": map[string]any{"kind": "endpoint", "id": "bob"},
|
||||
"body": map[string]any{"enc": "utf8", "data": "to-void"},
|
||||
"delay_ms": delay0,
|
||||
})
|
||||
if !sendResp.OK {
|
||||
t.Fatalf("send: %+v", sendResp)
|
||||
}
|
||||
msg := bob.WaitType(t, "msg", 8*time.Second)
|
||||
if msg["id"] != "dm-fatal-1" {
|
||||
t.Fatalf("bob msg=%v", msg)
|
||||
}
|
||||
|
||||
admin := adminHTTPClient(t, base)
|
||||
disableEP(t, admin, base, "bob")
|
||||
|
||||
fatal := bob.WaitType(t, "fatal", 8*time.Second)
|
||||
if fatal["reason"] != "disabled" {
|
||||
t.Fatalf("fatal=%v", fatal)
|
||||
}
|
||||
revoked := bob.WaitType(t, "revoked", 8*time.Second)
|
||||
if revoked["id"] != "dm-fatal-1" || revoked["reason"] != "endpoint_disabled" {
|
||||
t.Fatalf("revoked=%v", revoked)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUplinkResetPasswordFatal 验证重置登录密码后在线端收到 fatal(password_reset)。
|
||||
func TestUplinkResetPasswordFatal(t *testing.T) {
|
||||
dataDir := t.TempDir()
|
||||
cfgPath := writeTestConfig(t, dataDir)
|
||||
initAdminForTest(t, dataDir)
|
||||
enableRegistration(t, dataDir, "uplink-code")
|
||||
|
||||
cfg, err := config.Load(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if vErr := cfg.Validate(); vErr != nil {
|
||||
t.Fatal(vErr)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- runServe(ctx, cfg) }()
|
||||
defer func() {
|
||||
cancel()
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if err != nil {
|
||||
t.Errorf("serve exit: %v", err)
|
||||
}
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Error("serve did not stop")
|
||||
}
|
||||
}()
|
||||
|
||||
addr := waitListenAddr(t, dataDir, 15*time.Second)
|
||||
base := "http://" + addr
|
||||
|
||||
registerEP(t, base, "carol", "password12", "Carol")
|
||||
carol := mqttSessionLogin(t, base, "carol", "password12")
|
||||
defer carol.Close()
|
||||
|
||||
admin := adminHTTPClient(t, base)
|
||||
resetLoginPassword(t, admin, base, "carol", "password99xx")
|
||||
|
||||
fatal := carol.WaitType(t, "fatal", 8*time.Second)
|
||||
if fatal["reason"] != "password_reset" {
|
||||
t.Fatalf("fatal=%v", fatal)
|
||||
}
|
||||
}
|
||||
|
||||
func adminHTTPClient(t *testing.T, base string) *http.Client {
|
||||
t.Helper()
|
||||
jar, err := cookiejar.New(nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
client := &http.Client{Jar: jar, Timeout: 10 * time.Second}
|
||||
loginBody, _ := json.Marshal(map[string]string{
|
||||
"username": "admin",
|
||||
"password": "test-admin-password-xx",
|
||||
})
|
||||
resp, err := client.Post(base+"/api/admin/login", "application/json", bytes.NewReader(loginBody))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
_ = resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("admin login: %d %s", resp.StatusCode, raw)
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
func disableEP(t *testing.T, client *http.Client, base, id string) {
|
||||
t.Helper()
|
||||
req, err := http.NewRequest(http.MethodPatch, base+"/api/admin/endpoints/"+id,
|
||||
strings.NewReader(`{"enabled":false}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Nixmsg-Request", "1")
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
_ = resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("disable %s: %d %s", id, resp.StatusCode, raw)
|
||||
}
|
||||
}
|
||||
|
||||
func resetLoginPassword(t *testing.T, client *http.Client, base, id, password string) {
|
||||
t.Helper()
|
||||
body := `{"login_password":"` + password + `"}`
|
||||
req, err := http.NewRequest(http.MethodPost, base+"/api/admin/endpoints/"+id+"/reset-login-password",
|
||||
strings.NewReader(body))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Nixmsg-Request", "1")
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
_ = resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("reset password %s: %d %s", id, resp.StatusCode, raw)
|
||||
}
|
||||
}
|
||||
+36
-17
@@ -71,10 +71,6 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
loginLocks := auth.NewLoginLocks()
|
||||
memConns := message.NewMemoryConns()
|
||||
msgLim := message.LimitsFromFullConfig(cfg)
|
||||
metricsReg := metrics.New()
|
||||
db.Queue.OnBatchCommit = func(d time.Duration) {
|
||||
metricsReg.WriteCommitSeconds.Observe(d.Seconds())
|
||||
}
|
||||
|
||||
login := broker.NewLogin(broker.LoginOptions{
|
||||
DB: db,
|
||||
@@ -88,13 +84,11 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
msgApp := message.New(db, msgLim, hashPool,
|
||||
message.WithLocks(loginLocks),
|
||||
message.WithConnRegistry(memConns),
|
||||
message.WithMetrics(metricsReg),
|
||||
)
|
||||
uplink := &appUplink{
|
||||
msg: msgApp,
|
||||
conns: memConns,
|
||||
log: slog.Default(),
|
||||
metrics: metricsReg,
|
||||
msg: msgApp,
|
||||
conns: memConns,
|
||||
log: slog.Default(),
|
||||
}
|
||||
sess := broker.NewSession(broker.SessionOptions{
|
||||
Login: login,
|
||||
@@ -115,7 +109,6 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
Authenticator: login,
|
||||
Uplink: sess,
|
||||
Logger: slog.Default(),
|
||||
Metrics: metricsReg,
|
||||
OnPublishDropped: func(dropCtx context.Context, endpointID string, connID port.ConnID, payload []byte) {
|
||||
if dropErr := msgApp.OnPublishDropped(dropCtx, endpointID, connID, payload); dropErr != nil {
|
||||
slog.Error("on publish dropped", "endpoint", endpointID, "err", dropErr)
|
||||
@@ -132,7 +125,6 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
message.WithLocks(loginLocks),
|
||||
message.WithConnRegistry(memConns),
|
||||
message.WithDownlink(brk),
|
||||
message.WithMetrics(metricsReg),
|
||||
)
|
||||
uplink.msg = msgApp
|
||||
uplink.down = brk
|
||||
@@ -153,6 +145,7 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
MaxScheduleSeconds: int64(cfg.Limits.MaxScheduleSeconds),
|
||||
Logger: slog.Default(),
|
||||
ConnControl: brk,
|
||||
Downlink: brk,
|
||||
})
|
||||
uplink.identity = idApp
|
||||
|
||||
@@ -181,6 +174,7 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
Groups: groupApp,
|
||||
Config: cfg,
|
||||
Version: Version,
|
||||
// Kick：只断开，令牌不变，SDK 重连（PRD 踢下线）。
|
||||
KickEndpoint: func(kickCtx context.Context, endpointID string) (bool, error) {
|
||||
if _, found := brk.ConnInfoOf(endpointID); !found {
|
||||
return false, nil
|
||||
@@ -190,8 +184,37 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
}
|
||||
return true, nil
|
||||
},
|
||||
// 停用/删除/重置：先 fatal 再断开（DEVELOPMENT 6.8）。
|
||||
DisableKick: func(kickCtx context.Context, endpointID string) (bool, error) {
|
||||
if _, found := brk.ConnInfoOf(endpointID); !found {
|
||||
return false, nil
|
||||
}
|
||||
if err := sess.Disable(kickCtx, endpointID); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
},
|
||||
DeleteKick: func(kickCtx context.Context, endpointID string) (bool, error) {
|
||||
if _, found := brk.ConnInfoOf(endpointID); !found {
|
||||
return false, nil
|
||||
}
|
||||
if err := sess.Deleted(kickCtx, endpointID); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
},
|
||||
PasswordResetKick: func(kickCtx context.Context, endpointID string) (bool, error) {
|
||||
if _, found := brk.ConnInfoOf(endpointID); !found {
|
||||
return false, nil
|
||||
}
|
||||
if resetErr := sess.ResetPassword(kickCtx, endpointID); resetErr != nil {
|
||||
return false, resetErr
|
||||
}
|
||||
return true, nil
|
||||
},
|
||||
})
|
||||
|
||||
metricsReg := metrics.New()
|
||||
buildHandlers := func(proxies *listener.ProxySet) listener.Handlers {
|
||||
return listener.Handlers{
|
||||
MQTT: brk.WSHandler(proxies),
|
||||
@@ -273,7 +296,7 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
|
||||
loopCtx, loopCancel := context.WithCancel(ctx)
|
||||
defer loopCancel()
|
||||
go messageLoops(loopCtx, msgApp, memConns, db, hashPool, metricsReg)
|
||||
go messageLoops(loopCtx, msgApp, memConns)
|
||||
|
||||
<-ctx.Done()
|
||||
loopCancel()
|
||||
@@ -286,7 +309,7 @@ func runServe(ctx context.Context, cfg config.Config) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func messageLoops(ctx context.Context, msgApp *message.App, conns *message.MemoryConns, db *store.DB, hashPool auth.HashPool, met *metrics.Registry) {
|
||||
func messageLoops(ctx context.Context, msgApp *message.App, conns *message.MemoryConns) {
|
||||
t := time.NewTicker(time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
@@ -306,10 +329,6 @@ func messageLoops(ctx context.Context, msgApp *message.App, conns *message.Memor
|
||||
if err := msgApp.CleanupOnce(ctx, nowMs); err != nil {
|
||||
slog.Error("cleanup once", "err", err)
|
||||
}
|
||||
if err := metrics.SampleStoreGauges(ctx, met, db.Read); err != nil {
|
||||
slog.Debug("sample store gauges", "err", err)
|
||||
}
|
||||
metrics.SampleQueues(met, db.Queue.Len(), hashPool.QueueLen())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
"git.asio.asia/nixevol/NixMsg/internal/app/message"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/app/presence"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/metrics"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/protocol"
|
||||
)
|
||||
|
||||
@@ -24,7 +23,6 @@ type appUplink struct {
|
||||
conns *message.MemoryConns
|
||||
down port.Downlink
|
||||
log *slog.Logger
|
||||
metrics *metrics.Registry
|
||||
}
|
||||
|
||||
func (u *appUplink) OnSessionEstablished(ctx context.Context, conn port.ConnInfo) error {
|
||||
@@ -257,9 +255,6 @@ func (u *appUplink) replyErr(ctx context.Context, conn port.ConnInfo, rid, code,
|
||||
if rid == "" {
|
||||
rid = "0"
|
||||
}
|
||||
if u.metrics != nil && code != "" {
|
||||
u.metrics.ErrorsTotal.WithLabelValues(code).Inc()
|
||||
}
|
||||
resp := protocol.Resp{
|
||||
V: protocol.Version,
|
||||
Type: protocol.TypeResp,
|
||||
|
||||
+7
-7
@@ -1093,11 +1093,11 @@
|
||||
- 备选方案：broker 层对 Inline 发布做无锁队列。
|
||||
- 影响：`group_event` 可能略晚于 `resp` 到达；业务结果仍以 `resp` 为准。
|
||||
|
||||
### fix-issue-5
|
||||
### fix-issue-4
|
||||
|
||||
1. **指标在真实事件点打点，不新造名字**
|
||||
- 原条款：PRD F22 / DEVELOPMENT 4.3 / issue #5；DEVIATIONS P4 已定名但从未接线。
|
||||
- 实际做法：`nixmsg_connections{transport}` 在 broker `OnSessionEstablished`/`OnDisconnect` 末尾 Inc/Dec；`endpoints`/`deliveries_pending`/`messages_scheduled` 与写队列、哈希排队在 `messageLoops` 每秒按库/队列真实长度采样；`dispatch_to_push`/`ack` 直方图在成功推送与确认路径 Observe；写批提交耗时经 `store.Queue.OnBatchCommit`；`errors_total` 仅在上行 `replyErr` 时按错误码递增。门禁不变。
|
||||
- 原因：空指标等于监控未交付；采样避免在每条写路径上改大段分发逻辑，并减小与 #3/#6 的合并面。
|
||||
- 备选方案：全部改为纯事件加减（pending 等需在每处状态迁移维护计数）。
|
||||
- 影响：仪表盘按既有名字即可看在线连接与待投递；无对应事件时计数保持 0，不做假数。
|
||||
1. **接线补齐 Downlink 与停用/删除/重置密码 fatal**
|
||||
- 原条款：DEVELOPMENT 6.8 / 7.6：停用、删除、重置密码先发 `fatal` 再断开；已推送作废投递尽力发 `revoked`。
|
||||
- 实际做法：`serve` 给 `identity.New` 注入 `Downlink: brk`（作废后 `publishRevokes`）；`DisableKick`/`DeleteKick`/`PasswordResetKick` 分别接到 `Session.Disable`/`Deleted`/`ResetPassword`；`KickEndpoint` 仍只 `Kick`。Identity 在未接 Kick 钩子时仍可用 `ConnControl` 异步断开兜底。
|
||||
- 原因：原先 Downlink 未注入导致 revoked 丢失；管理路径只 `Kick`/`Disconnect` 不发 fatal。
|
||||
- 备选方案：仅在 identity 内 `PublishDown(fatal)` 再断开；联调中该路径不如 Session.fatalKick 稳，故生产致命踢线统一走 Session。
|
||||
- 影响：管理「踢下线」语义不变；SDK 可按 fatal 停止重连；接收方能收到已推送消息的 revoked。
|
||||
|
||||
@@ -102,6 +102,33 @@ func (h *Handler) kickEndpoint(ctx context.Context, id string) (bool, error) {
|
||||
return h.kick(ctx, id)
|
||||
}
|
||||
|
||||
func (h *Handler) passwordResetKick(ctx context.Context, id string) (bool, error) {
|
||||
if h.resetKick != nil {
|
||||
return h.resetKick(ctx, id)
|
||||
}
|
||||
return h.kickEndpoint(ctx, id)
|
||||
}
|
||||
|
||||
func (h *Handler) afterDisableKick(ctx context.Context, id string) {
|
||||
if h.disableKick != nil {
|
||||
_, _ = h.disableKick(ctx, id)
|
||||
return
|
||||
}
|
||||
if h.identity == nil {
|
||||
_, _ = h.kickEndpoint(ctx, id)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Handler) afterDeleteKick(ctx context.Context, id string) {
|
||||
if h.deleteKick != nil {
|
||||
_, _ = h.deleteKick(ctx, id)
|
||||
return
|
||||
}
|
||||
if h.identity == nil {
|
||||
_, _ = h.kickEndpoint(ctx, id)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Handler) handleEndpointList(w http.ResponseWriter, r *http.Request) {
|
||||
q := r.URL.Query()
|
||||
limit := defaultListLimit
|
||||
@@ -350,7 +377,7 @@ func (h *Handler) handleEndpointPatch(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
if !*req.Enabled {
|
||||
_, _ = h.kickEndpoint(r.Context(), id)
|
||||
h.afterDisableKick(r.Context(), id)
|
||||
}
|
||||
} else if req.Enabled != nil && !*req.Enabled && wasEnabled {
|
||||
_, _ = h.kickEndpoint(r.Context(), id)
|
||||
@@ -381,7 +408,7 @@ func (h *Handler) handleEndpointDelete(w http.ResponseWriter, r *http.Request) {
|
||||
httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
|
||||
return
|
||||
}
|
||||
_, _ = h.kickEndpoint(r.Context(), id)
|
||||
h.afterDeleteKick(r.Context(), id)
|
||||
h.audit(actorString(p), "endpoint_delete", id, "ok", ip)
|
||||
httpx.WriteOK(w, map[string]any{})
|
||||
}
|
||||
@@ -416,14 +443,14 @@ func (h *Handler) handleEndpointBatch(w http.ResponseWriter, r *http.Request) {
|
||||
case "disable":
|
||||
found, opErr = h.setEndpointEnabled(r.Context(), id, false)
|
||||
if found && opErr == nil {
|
||||
_, _ = h.kickEndpoint(r.Context(), id)
|
||||
h.afterDisableKick(r.Context(), id)
|
||||
}
|
||||
case "enable":
|
||||
found, opErr = h.setEndpointEnabled(r.Context(), id, true)
|
||||
case "delete":
|
||||
found, opErr = h.deleteEndpointBasic(r.Context(), id)
|
||||
if found && opErr == nil {
|
||||
_, _ = h.kickEndpoint(r.Context(), id)
|
||||
h.afterDeleteKick(r.Context(), id)
|
||||
}
|
||||
}
|
||||
if opErr != nil {
|
||||
@@ -512,7 +539,7 @@ func (h *Handler) handleEndpointResetLoginPassword(w http.ResponseWriter, r *htt
|
||||
httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在")
|
||||
return
|
||||
}
|
||||
_, _ = h.kickEndpoint(r.Context(), id)
|
||||
_, _ = h.passwordResetKick(r.Context(), id)
|
||||
h.audit(actorString(p), "endpoint_reset_login_password", id, "ok", ip)
|
||||
httpx.WriteOK(w, map[string]any{loginPasswordOnceKey: pw})
|
||||
}
|
||||
|
||||
+42
-30
@@ -41,6 +41,12 @@ type Deps struct {
|
||||
SecureCookies bool
|
||||
// KickEndpoint 踢下线钩子（只断开连接）；nil 时踢线为 no-op。
|
||||
KickEndpoint EndpointKickFunc
|
||||
// PasswordResetKick 重置登录密码后踢线（应发 fatal）；nil 时回退 KickEndpoint。
|
||||
PasswordResetKick EndpointKickFunc
|
||||
// DisableKick 停用后踢线（应发 fatal(disabled)）；nil 且已注入 Identity 时不再 Kick。
|
||||
DisableKick EndpointKickFunc
|
||||
// DeleteKick 删除后踢线（应发 fatal(deleted)）；nil 且已注入 Identity 时不再 Kick。
|
||||
DeleteKick EndpointKickFunc
|
||||
// Identity 端停用/启用/删除级联（I5）；nil 时回退为仅改 enabled/删行。
|
||||
Identity identity.Service
|
||||
|
||||
@@ -54,20 +60,23 @@ type Deps struct {
|
||||
|
||||
// Handler 是可挂载的管理接口（路由前缀 /api/admin/）。
|
||||
type Handler struct {
|
||||
db *store.DB
|
||||
hash auth.HashPool
|
||||
tokens auth.APITokens
|
||||
locks auth.LoginLocks
|
||||
log *slog.Logger
|
||||
trusted []*net.IPNet
|
||||
ttl time.Duration
|
||||
forceSec bool
|
||||
kick EndpointKickFunc
|
||||
identity identity.Service
|
||||
groups group.Service
|
||||
cfg config.Config
|
||||
version string
|
||||
startedAt time.Time
|
||||
db *store.DB
|
||||
hash auth.HashPool
|
||||
tokens auth.APITokens
|
||||
locks auth.LoginLocks
|
||||
log *slog.Logger
|
||||
trusted []*net.IPNet
|
||||
ttl time.Duration
|
||||
forceSec bool
|
||||
kick EndpointKickFunc
|
||||
resetKick EndpointKickFunc
|
||||
disableKick EndpointKickFunc
|
||||
deleteKick EndpointKickFunc
|
||||
identity identity.Service
|
||||
groups group.Service
|
||||
cfg config.Config
|
||||
version string
|
||||
startedAt time.Time
|
||||
|
||||
mux *http.ServeMux
|
||||
|
||||
@@ -96,22 +105,25 @@ func New(d Deps) *Handler {
|
||||
ver = "dev"
|
||||
}
|
||||
h := &Handler{
|
||||
db: d.DB,
|
||||
hash: d.Hash,
|
||||
tokens: d.Tokens,
|
||||
locks: d.Locks,
|
||||
log: d.Logger,
|
||||
trusted: d.TrustedProxies,
|
||||
ttl: ttl,
|
||||
forceSec: d.SecureCookies,
|
||||
kick: d.KickEndpoint,
|
||||
identity: d.Identity,
|
||||
groups: d.Groups,
|
||||
cfg: cfg,
|
||||
version: ver,
|
||||
startedAt: time.Now(),
|
||||
mux: http.NewServeMux(),
|
||||
lastUsed: make(map[string]time.Time),
|
||||
db: d.DB,
|
||||
hash: d.Hash,
|
||||
tokens: d.Tokens,
|
||||
locks: d.Locks,
|
||||
log: d.Logger,
|
||||
trusted: d.TrustedProxies,
|
||||
ttl: ttl,
|
||||
forceSec: d.SecureCookies,
|
||||
kick: d.KickEndpoint,
|
||||
resetKick: d.PasswordResetKick,
|
||||
disableKick: d.DisableKick,
|
||||
deleteKick: d.DeleteKick,
|
||||
identity: d.Identity,
|
||||
groups: d.Groups,
|
||||
cfg: cfg,
|
||||
version: ver,
|
||||
startedAt: time.Now(),
|
||||
mux: http.NewServeMux(),
|
||||
lastUsed: make(map[string]time.Time),
|
||||
}
|
||||
h.routes()
|
||||
return h
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/protocol"
|
||||
@@ -22,6 +23,9 @@ const (
|
||||
eventLeft = "left"
|
||||
eventMemberRemoved = "member_removed"
|
||||
eventDissolved = "dissolved"
|
||||
|
||||
// kickFlushDelay 给接线方 Session.Disable/Deleted 留出发 fatal 的窗口。
|
||||
kickFlushDelay = 20 * time.Millisecond
|
||||
)
|
||||
|
||||
type revokeItem struct {
|
||||
@@ -124,8 +128,13 @@ WHERE id = ?`, endpointID); e != nil {
|
||||
|
||||
a.publishRevokes(ctx, revokes)
|
||||
a.publishGroupEvents(ctx, notifies)
|
||||
// fatal+断开由 admin DisableKick/DeleteKick（Session.Disable/Deleted）完成。
|
||||
// 未接 Kick 钩子的单元测试仍可用 ConnControl 兜底断开。
|
||||
if a.connCtrl != nil {
|
||||
_ = a.connCtrl.Disconnect(ctx, endpointID, "", port.DisconnectFatal)
|
||||
go func() {
|
||||
time.Sleep(kickFlushDelay)
|
||||
_ = a.connCtrl.Disconnect(context.Background(), endpointID, "", port.DisconnectFatal)
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package identity_test
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/cookiejar"
|
||||
@@ -64,6 +65,82 @@ VALUES(?,?,?,?,0,0,1,?,?)`, id, id, "stub$login", nil, 1_700_000_000_000, 1_700_
|
||||
}
|
||||
}
|
||||
|
||||
func TestDisableEmitsRevokedForPushed(t *testing.T) {
|
||||
t.Parallel()
|
||||
dir := t.TempDir()
|
||||
db, err := store.Open(filepath.Join(dir, "data"), "FULL")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
fixed := time.UnixMilli(1_700_000_000_000)
|
||||
down := &message.RecordingDownlink{}
|
||||
ctrl := &port.StubConnControl{}
|
||||
idApp := identity.New(identity.Config{
|
||||
DB: db,
|
||||
Hash: auth.NewStubHashPool(),
|
||||
Locks: auth.NewStubLoginLocks(),
|
||||
Sessions: auth.NewSessionTokens(),
|
||||
MaxScheduleSeconds: int64(config.Default().Limits.MaxScheduleSeconds),
|
||||
Now: func() time.Time { return fixed },
|
||||
ConnControl: ctrl,
|
||||
Downlink: down,
|
||||
})
|
||||
ctx := context.Background()
|
||||
insertEPFull(t, db, "alice")
|
||||
insertEPFull(t, db, "bob")
|
||||
|
||||
err = db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
||||
res, e := tx.Exec(`
|
||||
INSERT INTO messages(
|
||||
id, sender_id, dest_kind, dest_id, meta, content_type, body_enc,
|
||||
send_at, keep, ttl_seconds, receipt, state, reason, created_at)
|
||||
VALUES('pushed-1','alice','endpoint','bob','{}','text/plain','utf8',?,1,0,0,'dispatched','',?)`,
|
||||
fixed.UnixMilli(), fixed.UnixMilli())
|
||||
if e != nil {
|
||||
return e
|
||||
}
|
||||
seq, _ := res.LastInsertId()
|
||||
_, e = tx.Exec(`
|
||||
INSERT INTO deliveries(seq, endpoint_id, send_at, keep, state, reason, updated_at, pushed_at, pushed_conn)
|
||||
VALUES(?,?,?,1,'pending','',?,?,?)`,
|
||||
seq, "bob", fixed.UnixMilli(), fixed.UnixMilli(), fixed.UnixMilli(), "c-bob")
|
||||
return e
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := idApp.Disable(ctx, "bob"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if down.FilterType(protocol.TypeRevoked) != 1 {
|
||||
t.Fatalf("want 1 revoked, got snapshots=%v", down.Snapshots())
|
||||
}
|
||||
p := down.Snapshots()[0]
|
||||
var head struct {
|
||||
Type string `json:"type"`
|
||||
Reason string `json:"reason"`
|
||||
ID string `json:"id"`
|
||||
}
|
||||
_ = json.Unmarshal(p.Payload, &head)
|
||||
if head.Type != protocol.TypeRevoked || head.Reason != "endpoint_disabled" || head.ID != "pushed-1" {
|
||||
t.Fatalf("revoked=%+v", head)
|
||||
}
|
||||
if p.EndpointID != "bob" || p.QoS != 1 {
|
||||
t.Fatalf("publish=%+v", p)
|
||||
}
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if len(ctrl.Calls) == 1 && ctrl.Calls[0] == "bob" {
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("disconnect calls=%v", ctrl.Calls)
|
||||
}
|
||||
|
||||
func TestF01DisableVoidsScheduledAndRejectsNew(t *testing.T) {
|
||||
t.Parallel()
|
||||
idApp, msgApp, db := openLifecycle(t)
|
||||
|
||||
@@ -17,8 +17,6 @@ func (a *App) Ack(ctx context.Context, endpointID string, req *protocol.Ack) (Ac
|
||||
nowMs := a.now().UnixMilli()
|
||||
var out AckResult
|
||||
var seq int64
|
||||
var ackLatencySec float64
|
||||
var observeAck bool
|
||||
err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
||||
err := tx.QueryRow(`SELECT seq FROM messages WHERE sender_id = ? AND id = ?`, req.From, req.ID).Scan(&seq)
|
||||
if err == sql.ErrNoRows {
|
||||
@@ -27,10 +25,6 @@ func (a *App) Ack(ctx context.Context, endpointID string, req *protocol.Ack) (Ac
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var pushedAt sql.NullInt64
|
||||
_ = tx.QueryRow(`
|
||||
SELECT pushed_at FROM deliveries
|
||||
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, seq, endpointID).Scan(&pushedAt)
|
||||
res, err := tx.Exec(`
|
||||
UPDATE deliveries SET state = ?, reason = '', pushed_conn = NULL, updated_at = ?
|
||||
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
|
||||
@@ -41,10 +35,6 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
|
||||
aff, _ := res.RowsAffected()
|
||||
if aff > 0 {
|
||||
out.Result = DeliveryAccepted
|
||||
if pushedAt.Valid && pushedAt.Int64 > 0 && nowMs >= pushedAt.Int64 {
|
||||
ackLatencySec = float64(nowMs-pushedAt.Int64) / 1000.0
|
||||
observeAck = true
|
||||
}
|
||||
if e := insertReceiptTx(tx, req.From, seq, endpointID, DeliveryAccepted, "", nowMs); e != nil {
|
||||
return e
|
||||
}
|
||||
@@ -65,9 +55,6 @@ SELECT state FROM deliveries WHERE seq = ? AND endpoint_id = ?`, seq, endpointID
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
if observeAck && a.met != nil {
|
||||
a.met.AckSeconds.Observe(ackLatencySec)
|
||||
}
|
||||
if out.Result == DeliveryAccepted {
|
||||
a.releaseLarge(seq, endpointID)
|
||||
}
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/auth"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/config"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/metrics"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/protocol"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/store"
|
||||
)
|
||||
@@ -87,7 +86,6 @@ type App struct {
|
||||
|
||||
down port.Downlink
|
||||
conns ConnRegistry
|
||||
met *metrics.Registry
|
||||
|
||||
mu sync.Mutex
|
||||
largeSem chan struct{}
|
||||
@@ -119,11 +117,6 @@ func WithConnRegistry(c ConnRegistry) Option {
|
||||
return func(a *App) { a.conns = c }
|
||||
}
|
||||
|
||||
// WithMetrics 注入 Prometheus 注册表（投递耗时直方图）。
|
||||
func WithMetrics(m *metrics.Registry) Option {
|
||||
return func(a *App) { a.met = m }
|
||||
}
|
||||
|
||||
// New 创建消息服务实现。
|
||||
func New(db *store.DB, lim Limits, hash auth.HashPool, opts ...Option) *App {
|
||||
if lim.RequestBurst <= 0 {
|
||||
|
||||
@@ -229,20 +229,11 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending' AND pushed_conn IS NULL`
|
||||
a.releaseLarge(it.seq, endpointID)
|
||||
}
|
||||
a.scheduleRepush(endpointID, time.Second)
|
||||
} else {
|
||||
a.observeDispatchToPush(it.sendAt, nowMs)
|
||||
}
|
||||
}
|
||||
return a.pushReceipts(ctx, endpointID, connID, nowMs)
|
||||
}
|
||||
|
||||
func (a *App) observeDispatchToPush(sendAtMs, pushedAtMs int64) {
|
||||
if a.met == nil || pushedAtMs < sendAtMs {
|
||||
return
|
||||
}
|
||||
a.met.DispatchToPushSeconds.Observe(float64(pushedAtMs-sendAtMs) / 1000.0)
|
||||
}
|
||||
|
||||
func (a *App) rejectTooLarge(ctx context.Context, seq int64, endpointID, senderID string, nowMs int64) error {
|
||||
return a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
||||
res, err := tx.Exec(`
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"time"
|
||||
|
||||
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
||||
"git.asio.asia/nixevol/NixMsg/internal/metrics"
|
||||
mqtt "github.com/mochi-mqtt/server/v2"
|
||||
"github.com/mochi-mqtt/server/v2/packets"
|
||||
)
|
||||
@@ -69,18 +68,15 @@ type Options struct {
|
||||
Logger *slog.Logger
|
||||
// OnPublishDropped 可选；nil 时仅打 debug 日志。
|
||||
OnPublishDropped PublishDroppedFunc
|
||||
// Metrics 可选；会话建立/断开时更新 nixmsg_connections。
|
||||
Metrics *metrics.Registry
|
||||
}
|
||||
|
||||
// Broker 内置 mochi，不自带监听端口。
|
||||
type Broker struct {
|
||||
server *mqtt.Server
|
||||
auth Authenticator
|
||||
uplink port.UplinkHandler
|
||||
log *slog.Logger
|
||||
onDrop PublishDroppedFunc
|
||||
metrics *metrics.Registry
|
||||
server *mqtt.Server
|
||||
auth Authenticator
|
||||
uplink port.UplinkHandler
|
||||
log *slog.Logger
|
||||
onDrop PublishDroppedFunc
|
||||
|
||||
hook *nixHook
|
||||
|
||||
@@ -109,7 +105,6 @@ type connState struct {
|
||||
handshook bool
|
||||
subscribedDown bool
|
||||
largeHeld int
|
||||
metricsCounted bool
|
||||
mu sync.Mutex
|
||||
|
||||
handshakeTimer *time.Timer
|
||||
@@ -156,7 +151,6 @@ func New(opts Options) (*Broker, error) {
|
||||
uplink: uplink,
|
||||
log: log,
|
||||
onDrop: opts.OnPublishDropped,
|
||||
metrics: opts.Metrics,
|
||||
current: make(map[string]*connState),
|
||||
byClient: make(map[*mqtt.Client]*connState),
|
||||
queues: make(map[string]*uplinkQueue),
|
||||
|
||||
@@ -190,7 +190,6 @@ func (h *nixHook) OnSessionEstablished(cl *mqtt.Client, _ packets.Packet) {
|
||||
MaxPacketSize: st.maxPacketSize,
|
||||
}
|
||||
_ = h.b.uplink.OnSessionEstablished(context.Background(), info)
|
||||
h.noteConnectionOpen(st)
|
||||
}
|
||||
|
||||
func (h *nixHook) OnDisconnect(cl *mqtt.Client, err error, _ bool) {
|
||||
@@ -230,27 +229,9 @@ func (h *nixHook) OnDisconnect(cl *mqtt.Client, err error, _ bool) {
|
||||
}
|
||||
if sess, ok := h.b.uplink.(*Session); ok {
|
||||
sess.HandleDisconnect(context.Background(), info, reason, isCurrent)
|
||||
h.noteConnectionClose(st)
|
||||
return
|
||||
}
|
||||
h.b.uplink.OnDisconnect(context.Background(), info, reason)
|
||||
h.noteConnectionClose(st)
|
||||
}
|
||||
|
||||
func (h *nixHook) noteConnectionOpen(st *connState) {
|
||||
if h.b.metrics == nil || st == nil || st.metricsCounted {
|
||||
return
|
||||
}
|
||||
h.b.metrics.Connections.WithLabelValues(string(st.transport)).Inc()
|
||||
st.metricsCounted = true
|
||||
}
|
||||
|
||||
func (h *nixHook) noteConnectionClose(st *connState) {
|
||||
if h.b.metrics == nil || st == nil || !st.metricsCounted {
|
||||
return
|
||||
}
|
||||
h.b.metrics.Connections.WithLabelValues(string(st.transport)).Dec()
|
||||
st.metricsCounted = false
|
||||
}
|
||||
|
||||
func (h *nixHook) OnQosComplete(cl *mqtt.Client, pk packets.Packet) {
|
||||
|
||||
@@ -1,118 +0,0 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.asio.asia/nixevol/NixMsg/internal/metrics"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
)
|
||||
|
||||
func TestConnectionMetricsIncDec(t *testing.T) {
|
||||
reg := metrics.New()
|
||||
b, err := New(Options{Authenticator: AllowAuthenticator{}, Metrics: reg})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _ = b.Close() }()
|
||||
|
||||
if got := gaugeValue(t, reg, "nixmsg_connections", "tcp"); got != 0 {
|
||||
t.Fatalf("before connect tcp=%v", got)
|
||||
}
|
||||
|
||||
r, w := net.Pipe()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
_ = b.AttachTCP(r)
|
||||
}()
|
||||
connectAndSubscribe(t, w, "ep-metrics", 0)
|
||||
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for {
|
||||
if _, ok := b.ConnInfoOf("ep-metrics"); ok {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("session not established")
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if got := gaugeValue(t, reg, "nixmsg_connections", "tcp"); got != 1 {
|
||||
t.Fatalf("after connect tcp=%v want 1", got)
|
||||
}
|
||||
|
||||
_ = w.Close()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("attach did not finish")
|
||||
}
|
||||
deadline = time.Now().Add(3 * time.Second)
|
||||
for {
|
||||
if got := gaugeValue(t, reg, "nixmsg_connections", "tcp"); got == 0 {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("after disconnect tcp=%v want 0", gaugeValue(t, reg, "nixmsg_connections", "tcp"))
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWSConnectionMetrics(t *testing.T) {
|
||||
reg := metrics.New()
|
||||
b, err := New(Options{Authenticator: AllowAuthenticator{}, Metrics: reg})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _ = b.Close() }()
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/mqtt", b.WSHandler(nil))
|
||||
srv := httptest.NewServer(mux)
|
||||
defer srv.Close()
|
||||
|
||||
// 仅验证 handler 暴露指标文本仍含初始标签；建连用 TCP 测即可。
|
||||
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.Handler().ServeHTTP(rec, req)
|
||||
body, _ := io.ReadAll(rec.Body)
|
||||
if !strings.Contains(string(body), `nixmsg_connections{transport="ws"} 0`) {
|
||||
t.Fatalf("missing ws series: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
func gaugeValue(t *testing.T, reg *metrics.Registry, name, transport string) float64 {
|
||||
t.Helper()
|
||||
mfs, err := reg.Gatherer().Gather()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, mf := range mfs {
|
||||
if mf.GetName() != name {
|
||||
continue
|
||||
}
|
||||
for _, m := range mf.GetMetric() {
|
||||
if matchLabel(m, "transport", transport) {
|
||||
return m.GetGauge().GetValue()
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Fatalf("metric %s transport=%s not found", name, transport)
|
||||
return 0
|
||||
}
|
||||
|
||||
func matchLabel(m *dto.Metric, key, val string) bool {
|
||||
for _, lp := range m.GetLabel() {
|
||||
if lp.GetName() == key && lp.GetValue() == val {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -1,36 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
)
|
||||
|
||||
// SampleStoreGauges 按库内真实计数刷新端总数、待投递、定时消息（无对应行则为 0）。
|
||||
func SampleStoreGauges(ctx context.Context, r *Registry, db *sql.DB) error {
|
||||
if r == nil || db == nil {
|
||||
return nil
|
||||
}
|
||||
var endpoints, pending, scheduled int
|
||||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM endpoints`).Scan(&endpoints); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM deliveries WHERE state = 'pending'`).Scan(&pending); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM messages WHERE state = 'scheduled'`).Scan(&scheduled); err != nil {
|
||||
return err
|
||||
}
|
||||
r.EndpointsTotal.Set(float64(endpoints))
|
||||
r.DeliveriesPending.Set(float64(pending))
|
||||
r.MessagesScheduled.Set(float64(scheduled))
|
||||
return nil
|
||||
}
|
||||
|
||||
// SampleQueues 刷新写队列与密码哈希排队长度（传入当前真实长度，不做估算）。
|
||||
func SampleQueues(r *Registry, writeQueueLen, passwordHashQueueLen int) {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.WriteQueueLength.Set(float64(writeQueueLen))
|
||||
r.PasswordHashQueue.Set(float64(passwordHashQueueLen))
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.asio.asia/nixevol/NixMsg/internal/store"
|
||||
)
|
||||
|
||||
func TestSampleStoreGaugesPendingNonZero(t *testing.T) {
|
||||
t.Parallel()
|
||||
dir := t.TempDir()
|
||||
db, err := store.Open(filepath.Join(dir, "data"), "FULL")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
ctx := context.Background()
|
||||
_, err = db.Write.ExecContext(ctx, `
|
||||
INSERT INTO endpoints(id, name, login_hash, enabled, source, created_at)
|
||||
VALUES ('alice', 'A', 'x', 1, 'admin', 1)`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = db.Write.ExecContext(ctx, `
|
||||
INSERT INTO messages(sender_id, id, dest_kind, dest_id, send_at, keep, ttl_seconds, receipt, state, reason, created_at, content_type, body_enc, meta)
|
||||
VALUES ('alice', 'm1', 'endpoint', 'bob', 100, 1, 0, 0, 'dispatched', '', 1, 'text/plain', 'utf8', '{}')`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var seq int64
|
||||
if err := db.Write.QueryRowContext(ctx, `SELECT seq FROM messages WHERE id='m1'`).Scan(&seq); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = db.Write.ExecContext(ctx, `
|
||||
INSERT INTO deliveries(seq, endpoint_id, send_at, keep, state, reason, attempts, updated_at)
|
||||
VALUES (?, 'bob', 100, 1, 'pending', '', 0, 1)`, seq)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = db.Write.ExecContext(ctx, `
|
||||
INSERT INTO messages(sender_id, id, dest_kind, dest_id, send_at, keep, ttl_seconds, receipt, state, reason, created_at, content_type, body_enc, meta)
|
||||
VALUES ('alice', 'm2', 'endpoint', 'bob', 999999, 0, 0, 0, 'scheduled', '', 1, 'text/plain', 'utf8', '{}')`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
reg := New()
|
||||
if err := SampleStoreGauges(ctx, reg, db.Read); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
SampleQueues(reg, 3, 1)
|
||||
|
||||
mfs, err := reg.Gatherer().Gather()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := map[string]float64{}
|
||||
for _, mf := range mfs {
|
||||
switch mf.GetName() {
|
||||
case "nixmsg_endpoints", "nixmsg_deliveries_pending", "nixmsg_messages_scheduled",
|
||||
"nixmsg_write_queue_length", "nixmsg_password_hash_queue_length":
|
||||
if len(mf.GetMetric()) > 0 {
|
||||
got[mf.GetName()] = mf.GetMetric()[0].GetGauge().GetValue()
|
||||
}
|
||||
}
|
||||
}
|
||||
if got["nixmsg_endpoints"] != 1 {
|
||||
t.Fatalf("endpoints=%v", got["nixmsg_endpoints"])
|
||||
}
|
||||
if got["nixmsg_deliveries_pending"] != 1 {
|
||||
t.Fatalf("pending=%v", got["nixmsg_deliveries_pending"])
|
||||
}
|
||||
if got["nixmsg_messages_scheduled"] != 1 {
|
||||
t.Fatalf("scheduled=%v", got["nixmsg_messages_scheduled"])
|
||||
}
|
||||
if got["nixmsg_write_queue_length"] != 3 || got["nixmsg_password_hash_queue_length"] != 1 {
|
||||
t.Fatalf("queues=%v", got)
|
||||
}
|
||||
}
|
||||
@@ -43,9 +43,6 @@ type Queue struct {
|
||||
ready bool
|
||||
lastWriteErr error
|
||||
pending int
|
||||
|
||||
// OnBatchCommit 可选；每次合并提交成功后回调耗时（秒级指标用）。
|
||||
OnBatchCommit func(d time.Duration)
|
||||
}
|
||||
|
||||
// NewQueue 创建合并写入队列并启动写 goroutine。
|
||||
@@ -132,7 +129,6 @@ func (q *Queue) loop() {
|
||||
}
|
||||
|
||||
func (q *Queue) runBatch(batch []writeJob) {
|
||||
started := time.Now()
|
||||
defer func() {
|
||||
q.mu.Lock()
|
||||
q.pending -= len(batch)
|
||||
@@ -233,9 +229,6 @@ func (q *Queue) runBatch(batch []writeJob) {
|
||||
}
|
||||
return
|
||||
}
|
||||
if q.OnBatchCommit != nil {
|
||||
q.OnBatchCommit(time.Since(started))
|
||||
}
|
||||
for _, o := range outcomes {
|
||||
if o.success {
|
||||
o.job.res <- nil
|
||||
|
||||
Reference in New Issue
Block a user