weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(service): run the event dispatcher as an independent cmd/events process

Browse Source 
HTTP Runtime does not start the worker. Each compose environment gets its own events service, health port, and bounded pool.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-15 09:08:15 +00:00
co-authored by Cursor
parent d5968ad4ba
commit 6bee7630bd
6 changed files with 224 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.doc/system-design/module-boundaries.yaml
+8
View File
@@ -239,6 +239,14 @@ table_owners:
- events
event_service_principals:
- events
event_delivery_attempts:
- events
event_consumer_modes:
- events
event_consumer_checkpoints:
- events
event_reconciliations:
- events
final_artifacts:
- artifactvault
artifact_payload_staging:
cmd/events/main.go
+127
View File
@@ -0,0 +1,127 @@
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"chat-agent-system/internal/config"
"chat-agent-system/internal/db"
"chat-agent-system/internal/events"
"github.com/jackc/pgx/v5/pgxpool"
)
func main() {
envPath := flag.String("env", ".env", "环境配置文件路径")
flag.Parse()
cfg, err := config.Load(*envPath)
if err != nil {
fmt.Fprintf(os.Stderr, "events: load config: %v\n", os.ErrInvalid)
os.Exit(1)
}
port := cfg.HTTPPort
if v := os.Getenv("EVENTS_HTTP_PORT"); v != "" {
if p, perr := parsePort(v); perr == nil {
port = p
}
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
pc, err := pgxpool.ParseConfig(cfg.ContextDSN())
if err != nil {
fmt.Fprintf(os.Stderr, "events: dsn\n")
os.Exit(1)
}
pc.MaxConns = 8
pool, err := pgxpool.NewWithConfig(ctx, pc)
if err != nil {
fmt.Fprintf(os.Stderr, "events: pool\n")
os.Exit(1)
}
defer pool.Close()
if err := db.RunMigrations(ctx, pool, db.ResolveMigrationsDir()); err != nil {
fmt.Fprintf(os.Stderr, "events: migrate\n")
os.Exit(1)
}
owner := fmt.Sprintf("events-%d", os.Getpid())
disp := events.NewDispatcher(pool, cfg.Env, owner)
mux := http.NewServeMux()
mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "env": cfg.Env, "owner": owner})
})
mux.HandleFunc("GET /ready", func(w http.ResponseWriter, r *http.Request) {
if err := events.PingPool(r.Context(), pool); err != nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "DB"})
return
}
if err := events.SchemaReady(r.Context(), pool); err != nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "SCHEMA"})
return
}
reg := events.NewRegistry()
if err := reg.Register(events.ConsumerNotify); err != nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "REGISTRY"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
})
mux.HandleFunc("GET /live", func(w http.ResponseWriter, _ *http.Request) {
if !disp.Alive(30 * time.Second) {
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "stale"})
return
}
writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
})
mux.HandleFunc("GET /metrics", func(w http.ResponseWriter, _ *http.Request) {
m := disp.Metrics
writeJSON(w, http.StatusOK, map[string]any{
"claimed": m.Claimed.Load(), "acked": m.Acked.Load(), "failed": m.Failed.Load(),
"fenced": m.Fenced.Load(), "dead": m.Dead.Load(), "quarantine": m.Quarantine.Load(),
"heartbeat": m.Heartbeat.Load(),
})
})
srv := &http.Server{Addr: fmt.Sprintf("%s:%d", cfg.HTTPHost, port), Handler: mux}
go func() {
consumers := []string{events.ConsumerNotify, events.ConsumerProjection, events.ConsumerWorkflowAdvance}
_ = disp.Serve(ctx, consumers)
}()
go func() {
<-ctx.Done()
shut, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
disp.StopClaim()
_ = srv.Shutdown(shut)
}()
fmt.Printf("events worker env=%s listen=%s:%d owner=%s mode=cluster-table\n", cfg.Env, cfg.HTTPHost, port, owner)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
os.Exit(1)
}
}
func parsePort(v string) (int, error) {
var p int
_, err := fmt.Sscanf(v, "%d", &p)
if err != nil || p <= 0 {
return 0, os.ErrInvalid
}
return p, nil
}
func writeJSON(w http.ResponseWriter, status int, body map[string]any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(body)
}
deploy/app/Dockerfile
+2
View File
@@ -18,6 +18,7 @@ COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /server ./cmd/server
RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /events ./cmd/events
FROM debian:bookworm-slim
@@ -28,6 +29,7 @@ RUN set -eux; \
WORKDIR /app
COPY --from=go-builder /server /app/server
COPY --from=go-builder /events /app/events
COPY --from=web-builder /web/dist /app/web/dist
COPY deploy/postgres/goose /app/deploy/postgres/goose
# Official markdown must be on disk at /app/templates/docs and /app/configs.
docker-compose.dev.yml
+29
View File
@@ -143,6 +143,35 @@ services:
- dev-net
- shared-net
events:
build:
context: .
dockerfile: deploy/app/Dockerfile
container_name: shuimu-agent-events-dev
entrypoint: ["/app/events"]
command: ["-env", "/dev/null"]
depends_on:
system-db:
condition: service_healthy
environment:
ENV: dev
HTTP_HOST: "0.0.0.0"
HTTP_PORT: "${EVENTS_HTTP_PORT:-8088}"
EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8088}"
DATA_DIR: /app/data
APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:-dev_secret_db_pass}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_dev}?sslmode=disable
ports:
- "127.0.0.1:${EVENTS_HTTP_PORT:-8088}:${EVENTS_HTTP_PORT:-8088}"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8088}/health"]
interval: 10s
timeout: 5s
retries: 6
start_period: 15s
restart: unless-stopped
networks:
- dev-net
networks:
dev-net:
name: chat-agent-dev-net
docker-compose.main.yml
+29
View File
@@ -142,6 +142,35 @@ services:
- main-net
- shared-net
events:
build:
context: .
dockerfile: deploy/app/Dockerfile
container_name: shuimu-agent-events-main
entrypoint: ["/app/events"]
command: ["-env", "/dev/null"]
depends_on:
system-db:
condition: service_healthy
environment:
ENV: main
HTTP_HOST: "0.0.0.0"
HTTP_PORT: "${EVENTS_HTTP_PORT:-8090}"
EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8090}"
DATA_DIR: /app/data
APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:?SYSTEM_DB_PASSWORD is required for main/production}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_main}?sslmode=disable
ports:
- "127.0.0.1:${EVENTS_HTTP_PORT:-8090}:${EVENTS_HTTP_PORT:-8090}"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8090}/health"]
interval: 10s
timeout: 5s
retries: 6
start_period: 15s
restart: always
networks:
- main-net
networks:
main-net:
name: chat-agent-main-net
docker-compose.test.yml
+29
View File
@@ -141,6 +141,35 @@ services:
- test-net
- shared-net
events:
build:
context: .
dockerfile: deploy/app/Dockerfile
container_name: shuimu-agent-events-test
entrypoint: ["/app/events"]
command: ["-env", "/dev/null"]
depends_on:
system-db:
condition: service_healthy
environment:
ENV: test
HTTP_HOST: "0.0.0.0"
HTTP_PORT: "${EVENTS_HTTP_PORT:-8089}"
EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8089}"
DATA_DIR: /app/data
APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:-test_secret_db_pass}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_test}?sslmode=disable
ports:
- "127.0.0.1:${EVENTS_HTTP_PORT:-8089}:${EVENTS_HTTP_PORT:-8089}"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8089}/health"]
interval: 10s
timeout: 5s
retries: 6
start_period: 15s
restart: unless-stopped
networks:
- test-net
networks:
test-net:
name: chat-agent-test-net