update 2026-10-07 18:47:42

This commit is contained in:
action committed 2026-10-07 18:47:42 +08:00
1 parent 617f7ca2a6
commit bd061545ed
4832 files changed
+1693456

No files matched your search

+151
View File
@@ -0,0 +1,151 @@
# SPDX-License-Identifier: GPL-2.0-only
#
# Copyright (C) 2023 ImmortalWrt.org
include $(TOPDIR)/rules.mk
PKG_NAME:=daed
PKG_VERSION:=2026.09.24
PKG_RELEASE:=2
PKG_SOURCE:=daed-src-2026.09.24-527bb166b1ae.tar.gz
PKG_SOURCE_URL:=https://github.com/kenzok8/openwrt-daede/releases/download/daed-src
PKG_SOURCE_SUBDIR:=$(PKG_NAME)-$(PKG_VERSION)
PKG_HASH:=527bb166b1aea6a54330ebbc8be5b0418baed74636b838aeaa78c17b70f2f996
PKG_LICENSE:=AGPL-3.0-only MIT
PKG_LICENSE_FILES:=LICENSE wing/LICENSE
PKG_MAINTAINER:=kenzok8
DAED_BUILD_DIR:=$(BUILD_DIR)/$(PKG_NAME)-$(PKG_VERSION)
PKG_BUILD_DIR:=$(DAED_BUILD_DIR)/wing
PKG_BUILD_DEPENDS:=golang/host bpf-headers
PKG_BUILD_PARALLEL:=1
PKG_BUILD_FLAGS:=no-mips16
# armv7 lacks a CO-RE vmlinux.h for the trace eBPF (under wing/dae-core), so
# trace's bpf2go fails there. Apply sbwml/openwrt_helloworld's arm patches
# (add vmlinux-arm.h + drop the kprobe that doesn't work) to build on arm.
ifeq ($(ARCH),arm)
PATCH_DIR:=$(CURDIR)/patches_arm
endif
GO_PKG:=github.com/daeuniverse/dae-wing
GO_PKG_LDFLAGS:= \
-s -w -buildid= \
-linkmode external -extldflags '-static -Wl,-s' \
-X '$(GO_PKG)/db.AppDescription=$(PKG_NAME) is a integration solution of dae, API and UI.'
GO_PKG_LDFLAGS_X:= \
$(GO_PKG)/db.AppName=$(PKG_NAME) \
$(GO_PKG)/db.AppVersion=$(PKG_VERSION)
GO_PKG_TAGS:=embedallowed,trace,timetzdata
include $(INCLUDE_DIR)/package.mk
include $(INCLUDE_DIR)/bpf.mk
include $(TOPDIR)/feeds/packages/lang/golang/golang-package.mk
GO_PKG_BUILD_VARS+= GOFLAGS="-trimpath -buildvcs=false -pgo=auto"
GO_PKG_TARGET_VARS+= \
CGO_LDFLAGS="$(TARGET_LDFLAGS) -static -Wl,-s" \
GOEXPERIMENT=newinliner,simd
define Package/daed/Default
SECTION:=net
CATEGORY:=Network
SUBMENU:=Web Servers/Proxies
URL:=https://github.com/daeuniverse/daed
endef
define Package/daed
$(call Package/daed/Default)
TITLE:=A Modern Dashboard For dae
DEPENDS:=$(GO_ARCH_DEPENDS) \
+ca-bundle +kmod-sched-core +kmod-sched-bpf \
+kmod-veth +v2ray-geoip +v2ray-geosite \
+@KERNEL_XDP_SOCKETS \
+DAED_USE_VMLINUX_BTF:vmlinux-btf
endef
define Package/daed/config
choice
prompt "BTF source for CO-RE"
default DAED_USE_KERNEL_BTF
depends on PACKAGE_daed
config DAED_USE_KERNEL_BTF
bool "Use kernel BTF (integrated)"
depends on KERNEL_DEBUG_INFO_BTF
config DAED_USE_VMLINUX_BTF
bool "Use vmlinux-btf package"
endchoice
endef
define Package/daed/description
daed is a backend of dae, provides a method to bundle arbitrary
frontend, dae and geodata into one binary.
endef
define Package/daed/conffiles
/etc/daed/wing.db
/etc/config/daed
endef
define Build/Prepare
mkdir -p $(DAED_BUILD_DIR)
$(TAR) --strip-components=1 -C $(DAED_BUILD_DIR) -xzf $(DL_DIR)/$(PKG_SOURCE)
$(if $(wildcard $(CURDIR)/patches/*.patch),$(call PatchDir,$(PKG_BUILD_DIR),$(CURDIR)/patches,))
$(if $(filter arm,$(ARCH)),$(if $(wildcard $(CURDIR)/patches_arm/*.patch),$(call PatchDir,$(PKG_BUILD_DIR),$(CURDIR)/patches_arm,)))
endef
DAE_CFLAGS:= \
-O2 -Wall -Werror \
-DMAX_MATCH_SET_LEN=1024 \
-I$(BPF_HEADERS_DIR)/tools/lib \
-I$(BPF_HEADERS_DIR)/arch/$(BPF_KARCH)/include/asm/mach-generic
define Build/Compile
( \
pushd $(PKG_BUILD_DIR) ; \
export \
$(GO_GENERAL_BUILD_CONFIG_VARS) \
$(GO_PKG_BUILD_CONFIG_VARS) \
$(GO_PKG_BUILD_VARS) ; \
go generate ./... ; \
cd dae-core ; \
export \
BPF_CLANG="$(CLANG)" \
BPF_STRIP_FLAG="-strip=$(LLVM_STRIP)" \
BPF_CFLAGS="$(DAE_CFLAGS)" \
BPF_TARGET="bpfel,bpfeb" \
BPF_TRACE_TARGET="$(GO_ARCH)" ; \
go generate ./control/control.go ; \
go generate ./trace/trace.go ; \
popd ; \
)
$(call GoPackage/Build/Compile)
endef
define Package/daed/install
$(call GoPackage/Package/Install/Bin,$(PKG_INSTALL_DIR))
$(INSTALL_DIR) $(1)/usr/bin
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/dae-wing $(1)/usr/bin/daed
$(INSTALL_BIN) $(CURDIR)/files/daed-guard $(1)/usr/bin/daed-guard
$(INSTALL_DIR) $(1)/usr/share/daed
$(INSTALL_DATA) $(CURDIR)/files/daed-cleanup.sh $(1)/usr/share/daed/cleanup.sh
$(INSTALL_DIR) $(1)/etc/config
$(INSTALL_CONF) $(CURDIR)/files/daed.config $(1)/etc/config/daed
$(INSTALL_DIR) $(1)/etc/init.d
$(INSTALL_BIN) $(CURDIR)/files/daed.init $(1)/etc/init.d/daed
$(INSTALL_DIR) $(1)/lib/upgrade/keep.d
$(INSTALL_DATA) $(CURDIR)/files/daed.keep $(1)/lib/upgrade/keep.d/daed
endef
$(eval $(call GoBinPackage,daed))
$(eval $(call BuildPackage,daed))
+106
View File
@@ -0,0 +1,106 @@
#!/bin/sh
# daed-cleanup.sh — reaper for stale daed kernel state.
#
# Removes:
# 1. Processes inside the `daens` netns (SIGTERM, then SIGKILL after 1s)
# 2. The `daens` network namespace itself (ip netns del, with
# umount+rm as fallback for zombie nsfs mounts)
# 3. The `dae0` veth pair in the host netns
#
# daed itself owns TC clsact detach. This opkg-side helper only
# reaps stale daens/dae0 state and never removes /sys/fs/bpf/daed.
# The pin directory is created during normal startup, so it is not
# a reliable leak indicator and must not block service lifecycle.
#
# The function stays sourceable by both init.d/daed and daed-guard.
daed_process_probe() {
if command -v pgrep >/dev/null 2>&1; then
pgrep -f '^/usr/bin/daed([[:space:]]|$)' >/dev/null 2>&1
case "$?" in
0) return 0 ;;
1) return 1 ;;
esac
fi
if command -v pidof >/dev/null 2>&1; then
pidof daed >/dev/null 2>&1
case "$?" in
0) return 0 ;;
1) return 1 ;;
esac
fi
return 2
}
daed_cleanup_runtime() {
local pid rc=0 probe_rc
# If daed userspace is currently running, do not touch the
# netns, the veth, or the eBPF dataplane. They are in active
# use; removing them would make every connection through dae
# hang.
daed_process_probe
probe_rc=$?
case "$probe_rc" in
0)
if [ "${DAED_GUARD_CLEANUP:-start}" = "start" ]; then
logger -t daed-init "cleanup: pre-start skipped because /usr/bin/daed is still running; refusing a second instance"
return 1
fi
logger -t daed-init "cleanup: post-exit skipped because another /usr/bin/daed instance is still running"
return 0
;;
1) ;;
*)
logger -t daed-init "cleanup: cannot determine whether daed is running; refusing to remove netns/veth"
return 1
;;
esac
# 1. Kill processes inside daens.
for pid in $(ip netns pids daens 2>/dev/null); do
kill "$pid" 2>/dev/null
done
if [ -n "$(ip netns pids daens 2>/dev/null)" ]; then
sleep 1
for pid in $(ip netns pids daens 2>/dev/null); do
kill -9 "$pid" 2>/dev/null
done
fi
# 2. Remove the daens netns. ip netns del can fail if a
# process still references it via /proc/<pid>/ns/net or
# because the umount has already happened. Try the
# umount/rm fallback.
if ! ip netns del daens 2>/dev/null; then
umount -l /run/netns/daens 2>/dev/null
if [ -e /run/netns/daens ] && ! rm -f /run/netns/daens 2>/dev/null; then
logger -t daed-init "cleanup: failed to remove /run/netns/daens (resource busy); a reboot may be required"
rc=1
fi
fi
# 3. Remove the dae0 veth pair.
if ip link show dae0 >/dev/null 2>&1; then
if ! ip link del dae0 2>/dev/null; then
logger -t daed-init "cleanup: failed to remove dae0 veth pair"
rc=1
fi
fi
# 4. The pin root is normal persistent state. Never remove it or
# make its existence change the cleanup result.
if [ -e /sys/fs/bpf/daed ]; then
logger -t daed-init "cleanup: /sys/fs/bpf/daed exists; leaving normal pin root unchanged"
fi
# Final verification covers only netns and veth state.
[ ! -e /run/netns/daens ] || return 1
! ip netns list 2>/dev/null | grep -Eq '^daens([[:space:]]|$)' || return 1
! ip link show dae0 >/dev/null 2>&1 || return 1
return $rc
}
+131
View File
@@ -0,0 +1,131 @@
#!/bin/sh
# Keep daed as a child so signals can be forwarded and stale netns/veth
# state can be cleaned before start and after exit. daed owns TC detach;
# /sys/fs/bpf/daed is normal persistent state and is never removed here.
. /usr/share/daed/cleanup.sh
# Pre-start cleanup refuses to remove runtime state while another
# daed instance is active, and fails closed if that probe is broken.
DAED_GUARD_CLEANUP=start
if ! daed_cleanup_runtime; then
echo "daed: stale /usr/bin/daed or netns state could not be verified or removed; refusing to start. Check process and netns state." >&2
logger -t daed-init "pre-start cleanup failed: refusing to start daed"
exit 1
fi
# Keep daed as a child so post-exit cleanup runs before procd can
# respawn it.
child_pid=
pending_signal=
shutdown_signal=
shutdown_elapsed=0
forced_kill=0
child_term_timeout=20
forward_signal() {
local sig="$1"
if [ -z "$child_pid" ]; then
pending_signal="$sig"
logger -t daed-init "signal $sig received before daed child started; launch cancelled"
return 0
fi
case "$sig" in
TERM|INT|QUIT)
if [ -z "$shutdown_signal" ]; then
shutdown_signal="$sig"
shutdown_elapsed=0
fi
;;
esac
kill -"$sig" "$child_pid" 2>/dev/null
}
exit_with_signal() {
local sig="$1"
trap - TERM INT HUP QUIT
kill -"$sig" "$$" 2>/dev/null
exit 1
}
child_is_running() {
local state
kill -0 "$child_pid" 2>/dev/null || return 1
state=$(awk '{ print $3 }' "/proc/$child_pid/stat" 2>/dev/null)
[ "$state" != "Z" ]
}
trap 'forward_signal TERM' TERM
trap 'forward_signal INT' INT
trap 'forward_signal HUP' HUP
trap 'forward_signal QUIT' QUIT
start_child() {
local sig
# Keep this check inside the function as well as at the call site:
# it closes the ordinary pre-start window, while the pending-signal
# path below handles a signal arriving during the background fork.
[ -z "$pending_signal" ] || return 125
/usr/bin/daed "$@" &
child_pid=$!
if [ -n "$pending_signal" ]; then
sig="$pending_signal"
pending_signal=
forward_signal "$sig"
fi
}
if [ -n "$pending_signal" ]; then
logger -t daed-init "refusing to start daed after pending signal $pending_signal"
exit_with_signal "$pending_signal"
fi
# Best-effort OOM preference; failure must not block startup.
if ! echo -16 > /proc/self/oom_score_adj 2>/dev/null; then
logger -t daed-init "warn: failed to set /proc/self/oom_score_adj; continuing without OOM preference"
fi
if ! start_child "$@"; then
logger -t daed-init "refusing to start daed after pending signal $pending_signal"
if [ -n "$pending_signal" ]; then
exit_with_signal "$pending_signal"
fi
exit 1
fi
status=0
reaped=0
while [ "$reaped" -eq 0 ]; do
if [ -n "$shutdown_signal" ] && child_is_running; then
if [ "$shutdown_elapsed" -ge "$child_term_timeout" ]; then
if [ "$forced_kill" -eq 0 ]; then
logger -t daed-init "daed did not exit within ${child_term_timeout}s after $shutdown_signal; sending KILL"
kill -KILL "$child_pid" 2>/dev/null
forced_kill=1
fi
else
sleep 1
shutdown_elapsed=$((shutdown_elapsed + 1))
continue
fi
sleep 1
continue
fi
wait "$child_pid" 2>/dev/null
status=$?
if ! child_is_running; then
reaped=1
fi
done
child_pid=
trap - TERM INT HUP QUIT
# Post-exit cleanup runs after the child is reaped.
DAED_GUARD_CLEANUP=post-exit
cleanup_status=0
daed_cleanup_runtime || cleanup_status=$?
if [ "$cleanup_status" -ne 0 ]; then
echo "daed: runtime cleanup after exit failed" >&2
logger -t daed-init "post-exit cleanup failed; check ip netns / ip link show"
fi
if [ "$status" -ne 0 ]; then
exit "$status"
fi
exit "$cleanup_status"
+9
View File
@@ -0,0 +1,9 @@
config daed 'config'
option enabled '0'
option listen_addr '0.0.0.0:2023'
option log_maxbackups '1'
option log_maxsize '5'
option subscribe_auto_update '0'
option subscribe_update_cycle 'daily'
option subscribe_update_hour '4'
+83
View File
@@ -0,0 +1,83 @@
#!/bin/sh /etc/rc.common
# Copyright (C) 2023 Tianling Shen <cnsztl@immortalwrt.org>
# daed-guard handles bounded child shutdown and post-exit netns/veth cleanup.
# Keep the log file and a pre-stop state snapshot for diagnostics.
USE_PROCD=1
START=99
CONF="daed"
PROG="/usr/bin/daed-guard"
LOG="/var/log/daed/daed.log"
. /usr/share/daed/cleanup.sh
log() {
logger -t daed-init "$@"
}
start_service() {
log "start: begin"
config_load "$CONF"
local enabled
config_get_bool enabled "config" "enabled" "0"
if [ "$enabled" -ne "1" ]; then
log "start: config disabled, exit"
return 1
fi
local listen_addr log_maxbackups log_maxsize
config_get listen_addr "config" "listen_addr" "0.0.0.0:2023"
config_get log_maxbackups "config" "log_maxbackups" "1"
config_get log_maxsize "config" "log_maxsize" "5"
log "start: listen=$listen_addr log_maxbackups=$log_maxbackups log_maxsize=$log_maxsize"
procd_open_instance "$CONF"
procd_set_param env DAE_LOCATION_ASSET="/usr/share/v2ray" TZ="$(uci -q get system.@system[0].zonename)"
procd_set_param command "$PROG" run
procd_append_param command --config "/etc/daed/"
procd_append_param command --listen "$listen_addr"
procd_append_param command --logfile "$LOG"
procd_append_param command --logfile-maxbackups "$log_maxbackups"
procd_append_param command --logfile-maxsize "$log_maxsize"
procd_set_param limits core="unlimited"
procd_set_param limits nofile="1000000 1000000"
# daed-guard escalates its child after 20 seconds. Leave time for
# reap and post-exit cleanup before procd kills the wrapper.
procd_set_param term_timeout 30
# procd_set_param respawn: arguments are (threshold, timeout, retry).
# threshold = runtime that resets the short-lived exit counter
# timeout = seconds to wait between retries
# retry = maximum short-lived exits before procd gives up
# Reset the counter after one hour of stable runtime; otherwise retry
# after 5 seconds and stop after 10 failures.
procd_set_param respawn 3600 5 10
# daed-guard sets oom_score_adj before forking; procd has no
# oom_adj/oom_score_adj parameter.
# procd_set_param stdout 1
procd_set_param stderr 1
procd_close_instance
log "start: procd_open_instance done"
}
stop_service() {
log "stop: begin"
# Cleanup runs in daed-guard after its child exits. Record only a
# pre-stop snapshot here; pin entries do not prove TC attachment.
local pinned="" ns_left=""
if [ -d /sys/fs/bpf/daed ]; then
pinned=$(ls /sys/fs/bpf/daed 2>/dev/null | tr '\n' ' ')
fi
if ip netns list 2>/dev/null | grep -q '^daens'; then
ns_left="daens"
fi
log "stop: pre-stop state — bpf_pin_entries=[${pinned:-none}] netns_left=[${ns_left:-none}]"
}
service_triggers() {
procd_add_reload_trigger "$CONF"
}
+1
View File
@@ -0,0 +1 @@
/etc/daed/wing.db-wal
@@ -0,0 +1,19 @@
From: kenzok8 <kenzok8@noreply>
Date: 2026-06-30
Subject: [PATCH] daed: raise manual latency probe concurrency to 16
Raise concurrent probes 8->16 so manual latency tests finish faster
when many nodes (incl. dead/timeout ones) are present.
Reported: https://github.com/kenzok8/openwrt-daede/issues/20
---
--- a/graphql/service/node/latency_mutation_utils.go
+++ b/graphql/service/node/latency_mutation_utils.go
@@ -19,7 +19,7 @@
"github.com/sirupsen/logrus"
)
-const latencyProbeConcurrency = 8
+const latencyProbeConcurrency = 16
func TestLatencies(ctx context.Context, ids *[]graphql.ID) ([]*LatencyResolver, error) {
option, err := latencyProbeOption(ctx)
@@ -0,0 +1,63 @@
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
index 7a3c33f..fd61554 100644
--- a/graphql/service/config/mutation_utils.go
+++ b/graphql/service/config/mutation_utils.go
@@ -9,11 +9,13 @@ import (
"context"
"errors"
"fmt"
+ "os"
"reflect"
"regexp"
"sort"
"strings"
"sync"
+ "time"
"github.com/daeuniverse/dae-wing/common"
"github.com/daeuniverse/dae-wing/dae"
@@ -258,6 +260,15 @@ func Rename(ctx context.Context, _id graphql.ID, name string) (n int32, err erro
var runLock sync.Mutex
+const reloadWaitTimeout = 75 * time.Second
+
+func restartAfterReloadTimeout() {
+ go func() {
+ time.Sleep(time.Second)
+ os.Exit(1)
+ }()
+}
+
func Run(d *gorm.DB, noLoad bool) (n int32, err error) {
if ok := runLock.TryLock(); !ok {
return 0, fmt.Errorf("the last request didn't complete; make a cup of tea and take a break")
@@ -455,13 +466,24 @@ func Run(d *gorm.DB, noLoad bool) (n int32, err error) {
/// Reload with current config.
chReloadCallback := make(chan error)
- dae.ChReloadConfigs <- &dae.ReloadMessage{
+ reloadMsg := &dae.ReloadMessage{
Config: c,
Callback: chReloadCallback,
}
- errReload := <-chReloadCallback
- if errReload != nil {
- return 0, fmt.Errorf("failed to load new config: %w; see more in log", errReload)
+ select {
+ case dae.ChReloadConfigs <- reloadMsg:
+ case <-time.After(reloadWaitTimeout):
+ restartAfterReloadTimeout()
+ return 0, fmt.Errorf("failed to submit reload within %s; restarting daed to recover runtime", reloadWaitTimeout)
+ }
+ select {
+ case errReload := <-chReloadCallback:
+ if errReload != nil {
+ return 0, fmt.Errorf("failed to load new config: %w; see more in log", errReload)
+ }
+ case <-time.After(reloadWaitTimeout):
+ restartAfterReloadTimeout()
+ return 0, fmt.Errorf("reload timed out after %s; restarting daed to recover runtime", reloadWaitTimeout)
}
// Save running status
@@ -0,0 +1,92 @@
From: kenzok8 <kenzok8@noreply>
Date: 2026-09-23
Subject: [PATCH] daed: bound startup network wait on new wing
Return an error after 60 seconds and retry on the next generation.
---
diff --git a/dae/run.go b/dae/run.go
index 7c9a5cc..7f00748 100644
--- a/dae/run.go
+++ b/dae/run.go
@@ -19,6 +19,7 @@ import (
"github.com/daeuniverse/dae/control"
"github.com/daeuniverse/dae/pkg/config_parser"
"github.com/daeuniverse/dae/pkg/logger"
+ "github.com/daeuniverse/outbound/netproxy"
"github.com/daeuniverse/outbound/protocol/direct"
"github.com/mohae/deepcopy"
"github.com/sirupsen/logrus"
@@ -43,7 +44,8 @@ var EmptyConfig *daeConfig.Config
// active is the generation currently serving traffic. It is nil before the
// first generation is ready and while a reload has retired the old one.
var active atomic.Pointer[control.ControlPlane]
-var onceWaitingNetwork sync.Once
+var waitingNetworkMu sync.Mutex
+var networkReady bool
func init() {
sections, err := config_parser.Parse(`global{} routing{}`)
@@ -64,6 +66,19 @@ func ControlPlane() (*control.ControlPlane, error) {
return c, nil
}
+func waitForNetwork(log *logrus.Logger, directDialer netproxy.Dialer) error {
+ waitingNetworkMu.Lock()
+ defer waitingNetworkMu.Unlock()
+ if networkReady {
+ return nil
+ }
+ if err := WaitForNetwork(log, directDialer); err != nil {
+ return err
+ }
+ networkReady = true
+ return nil
+}
+
// generation is one control plane with its listener. retired is set before
// the plane is torn down so a Serve error from a plane that is being replaced
// is not mistaken for the active datapath dying.
@@ -345,9 +360,9 @@ func newControlPlane(log *logrus.Logger, dnsCache map[string]*control.DnsCache,
if !conf.Global.DisableWaitingNetwork && len(conf.Global.WanInterface) > 0 {
// Wait for network for WAN ready.
- onceWaitingNetwork.Do(func() {
- WaitForNetwork(log, directDialers.Symmetric)
- })
+ if err = waitForNetwork(log, directDialers.Symmetric); err != nil {
+ return nil, err
+ }
}
/// Get subscription -> nodeList mapping.
diff --git a/dae/utils.go b/dae/utils.go
index 2b4f6bf..93efb7e 100644
--- a/dae/utils.go
+++ b/dae/utils.go
@@ -94,8 +94,9 @@ func preprocessWanInterfaceAuto(params *daeConfig.Config) error {
return nil
}
-func WaitForNetwork(log *logrus.Logger, directDialer netproxy.Dialer) {
+func WaitForNetwork(log *logrus.Logger, directDialer netproxy.Dialer) error {
epo := 5 * time.Second
+ deadline := time.Now().Add(60 * time.Second)
client := http.Client{
Transport: &http.Transport{
DialContext: func(ctx context.Context, network, addr string) (c net.Conn, err error) {
@@ -114,6 +115,9 @@ func WaitForNetwork(log *logrus.Logger, directDialer netproxy.Dialer) {
}
log.Infoln("Waiting for network...")
for i := 0; ; i++ {
+ if time.Now().After(deadline) {
+ return fmt.Errorf("network unavailable after 60 seconds")
+ }
resp, err := client.Get(cmd.CheckNetworkLinks[i%len(cmd.CheckNetworkLinks)])
if err != nil {
log.Debugln("CheckNetwork:", err)
@@ -133,4 +137,5 @@ func WaitForNetwork(log *logrus.Logger, directDialer netproxy.Dialer) {
time.Sleep(epo)
}
log.Infoln("Network online.")
+ return nil
}
@@ -0,0 +1,606 @@
From: kenzok8 <kenzok8@noreply>
Date: 2026-09-23
Subject: [PATCH] daed: strictly reconcile subscription updates
Reconcile nodes by link, preserve fixed-group nodes, and apply committed changes.
---
diff --git a/cmd/run.go b/cmd/run.go
index 7dce577..978f46c 100644
--- a/cmd/run.go
+++ b/cmd/run.go
@@ -161,11 +161,8 @@ func restoreRunningState() (err error) {
if !reload {
return nil
}
- tx := db.BeginTx(context.TODO())
// Reload.
- if _, err = config.Run(tx, false); err != nil {
- tx.Rollback()
-
+ if _, err = config.Run(context.TODO(), false); err != nil {
// Another tx.
// Set running = false.
tx2 := db.BeginTx(context.TODO())
@@ -183,7 +180,6 @@ func restoreRunningState() (err error) {
tx2.Commit()
return err
}
- tx.Commit()
return nil
}
diff --git a/graphql/mutation.go b/graphql/mutation.go
index af065fd..d462ecc 100644
--- a/graphql/mutation.go
+++ b/graphql/mutation.go
@@ -256,16 +256,7 @@ func (r *MutationResolver) SelectConfig(args *struct {
func (r *MutationResolver) Run(args *struct {
Dry bool
}) (int32, error) {
- tx := db.BeginTx(context.TODO())
- ret, err := config.Run(tx, args.Dry)
- if err != nil {
- tx.Rollback()
- return 0, err
- }
- if err = tx.Commit().Error; err != nil {
- return 0, fmt.Errorf("commit run: %w", err)
- }
- return ret, nil
+ return config.Run(context.TODO(), args.Dry)
}
func (r *MutationResolver) CreateDns(args *struct {
diff --git a/graphql/root_schema.go b/graphql/root_schema.go
index 8b82db6..a96cace 100644
--- a/graphql/root_schema.go
+++ b/graphql/root_schema.go
@@ -121,7 +121,7 @@ type Mutation {
# tagSubscription is to give the subscription a new tag.
tagSubscription(id: ID!, tag: String!): Int! @hasRole(role: ADMIN)
- # updateSubscription is to re-fetch subscription and resolve subscription into nodes. Old nodes that independently belong to any groups will not be removed.
+ # updateSubscription re-fetches and strictly reconciles subscription nodes. Stale nodes pinned by fixed groups become independent nodes.
updateSubscription(id: ID!): Subscription! @hasRole(role: ADMIN)
# updateSubscriptionLink is to update the subscription link without re-fetching nodes.
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
index e6bf9da..8ab756b 100644
--- a/graphql/service/config/mutation_utils.go
+++ b/graphql/service/config/mutation_utils.go
@@ -21,6 +21,7 @@ import (
"github.com/daeuniverse/dae-wing/dae"
"github.com/daeuniverse/dae-wing/db"
"github.com/daeuniverse/dae-wing/graphql/service/config/global"
+ "github.com/daeuniverse/dae-wing/graphql/service/general"
daeConfig "github.com/daeuniverse/dae/config"
"github.com/daeuniverse/dae/pkg/config_parser"
"github.com/graph-gophers/graphql-go"
@@ -269,21 +270,69 @@ func restartAfterReloadTimeout() {
}()
}
-func Run(d *gorm.DB, noLoad bool) (n int32, err error) {
+func runTransaction(ctx context.Context, noLoad bool) (n int32, err error) {
+ tx := db.BeginTx(ctx)
+ if tx.Error != nil {
+ return 0, tx.Error
+ }
+ n, err = runLocked(tx, noLoad)
+ if err != nil {
+ tx.Rollback()
+ return 0, err
+ }
+ if err = tx.Commit().Error; err != nil {
+ return 0, err
+ }
+ return n, nil
+}
+
+func Run(ctx context.Context, noLoad bool) (n int32, err error) {
if ok := runLock.TryLock(); !ok {
return 0, fmt.Errorf("the last request didn't complete; make a cup of tea and take a break")
}
defer runLock.Unlock()
+ return runTransaction(ctx, noLoad)
+}
+
+func ApplyIfRunning(ctx context.Context) error {
+ runLock.Lock()
+ defer runLock.Unlock()
+
+ daeResolver := general.DaeResolver{Ctx: ctx}
+ running, err := daeResolver.Running()
+ if err != nil || !running {
+ return err
+ }
+ modified, err := daeResolver.Modified()
+ if err != nil || !modified {
+ return err
+ }
+ _, err = runTransaction(ctx, false)
+ return err
+}
+
+func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
//// Dry run.
if noLoad {
ch := make(chan error)
- dae.ChReloadConfigs <- &dae.ReloadMessage{
+ reloadMsg := &dae.ReloadMessage{
Config: dae.EmptyConfig,
Callback: ch,
}
- err = <-ch
- if err != nil {
- return 0, fmt.Errorf("failed to dryrun: %w; see more in log and report bugs", err)
+ select {
+ case dae.ChReloadConfigs <- reloadMsg:
+ case <-time.After(reloadWaitTimeout):
+ restartAfterReloadTimeout()
+ return 0, fmt.Errorf("failed to submit dryrun within %s; restarting daed to recover runtime", reloadWaitTimeout)
+ }
+ select {
+ case err = <-ch:
+ if err != nil {
+ return 0, fmt.Errorf("failed to dryrun: %w; see more in log and report bugs", err)
+ }
+ case <-time.After(reloadWaitTimeout):
+ restartAfterReloadTimeout()
+ return 0, fmt.Errorf("dryrun timed out after %s; restarting daed to recover runtime", reloadWaitTimeout)
}
// Running -> false
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
index 7e6dc08..e87bf78 100644
--- a/graphql/service/subscription/mutation_utils.go
+++ b/graphql/service/subscription/mutation_utils.go
@@ -18,6 +18,7 @@ import (
"github.com/daeuniverse/dae-wing/dae"
"github.com/daeuniverse/dae-wing/db"
"github.com/daeuniverse/dae-wing/graphql/internal"
+ "github.com/daeuniverse/dae-wing/graphql/service/config"
"github.com/daeuniverse/dae-wing/graphql/service/node"
"github.com/daeuniverse/dae/common/subscription"
"github.com/go-co-op/gocron"
@@ -175,8 +176,9 @@ func AutoUpdateVersionByIds(d *gorm.DB, ids []uint) (err error) {
}
var (
- schedulerCache = make(map[uint]*gocron.Scheduler)
- schedulerMu sync.RWMutex
+ schedulerCache = make(map[uint]*gocron.Scheduler)
+ schedulerMu sync.RWMutex
+ subscriptionUpdateMu sync.Mutex
)
func UpdateAll(ctx context.Context) {
@@ -262,68 +264,161 @@ func Update(ctx context.Context, _id graphql.ID) (r *Resolver, err error) {
return &Resolver{Subscription: m}, nil
}
-func UpdateById(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
- // Fetch node links.
- var m db.Subscription
- if err = db.DB(ctx).Where(&db.Subscription{ID: subId}).First(&m).Error; err != nil {
- return nil, err
+func reconcileSubscriptionNodes(tx *gorm.DB, subId uint, links []string) error {
+ var existing []db.Node
+ if err := tx.Where("subscription_id = ?", subId).Find(&existing).Error; err != nil {
+ return err
}
- links, err := fetchLinks(m.Link)
- if err != nil {
- return nil, err
+
+ incoming := make(map[string]struct{}, len(links))
+ for _, link := range links {
+ incoming[link] = struct{}{}
}
- tx := db.BeginTx(ctx)
- defer func() {
- if err == nil {
- err = tx.Commit().Error
+ current := make(map[string]struct{}, len(existing))
+ var staleIds []uint
+ for _, n := range existing {
+ if _, ok := incoming[n.Link]; ok {
+ current[n.Link] = struct{}{}
+ continue
+ }
+ staleIds = append(staleIds, n.ID)
+ }
+
+ fixed := make(map[uint]struct{})
+ if len(staleIds) > 0 {
+ var fixedIds []uint
+ if err := tx.Table("group_nodes").
+ Distinct("group_nodes.node_id").
+ Joins("INNER JOIN groups ON groups.id = group_nodes.group_id").
+ Where("groups.policy = ?", "fixed").
+ Where("group_nodes.node_id IN ?", staleIds).
+ Pluck("group_nodes.node_id", &fixedIds).Error; err != nil {
+ return err
+ }
+ for _, id := range fixedIds {
+ fixed[id] = struct{}{}
+ }
+ }
+
+ var detachIds, deleteIds []uint
+ for _, id := range staleIds {
+ if _, ok := fixed[id]; ok {
+ detachIds = append(detachIds, id)
} else {
- tx.Rollback()
+ deleteIds = append(deleteIds, id)
+ }
+ }
+
+ if len(staleIds) > 0 {
+ if err := node.AutoUpdateVersionByIds(tx, staleIds); err != nil {
+ return err
+ }
+ }
+ if len(detachIds) > 0 {
+ if err := tx.Model(&db.Node{}).
+ Where("id IN ?", detachIds).
+ Update("subscription_id", nil).Error; err != nil {
+ return err
+ }
+ }
+ if len(deleteIds) > 0 {
+ if err := tx.Exec("DELETE FROM group_nodes WHERE node_id IN ?", deleteIds).Error; err != nil {
+ return err
+ }
+ if err := tx.Where("id IN ?", deleteIds).Delete(&db.Node{}).Error; err != nil {
+ return err
}
- }()
- // Remove those nodes whose subscription are independent from any groups.
- subQuery := tx.Raw(`select nodes.id as id
- from nodes
- inner join group_nodes on group_nodes.node_id = nodes.id
- where subscription_id = ?`, subId)
-
- if err = tx.Where("subscription_id = ?", subId).
- Where("id not in (?)", subQuery).
- Select(clause.Associations).
- Delete(&db.Node{}).Error; err != nil {
- return nil, err
}
- // Import node links.
+
+ seen := make(map[string]struct{}, len(links))
var args []*internal.ImportArgument
for _, link := range links {
+ if _, ok := seen[link]; ok {
+ continue
+ }
+ seen[link] = struct{}{}
+ if _, ok := current[link]; ok {
+ continue
+ }
args = append(args, &internal.ImportArgument{Link: link})
}
- result, err := node.Import(tx, false, &subId, args)
+ result, err := node.Import(tx, true, &subId, args)
if err != nil {
- return nil, err
+ return errors.New("failed to import subscription node")
}
- hasAnyCandidate := false
for _, r := range result {
- if r.Error == nil || *r.Error == node.DuplicatedError.Error() {
- hasAnyCandidate = true
- break
+ if r.Error != nil {
+ return errors.New("failed to import subscription node")
}
}
- if !hasAnyCandidate {
- return nil, fmt.Errorf("interrupt to update subscription: no any valid node can be imported")
+
+ var count int64
+ if err := tx.Model(&db.Node{}).Where("subscription_id = ?", subId).Count(&count).Error; err != nil {
+ return err
}
- // Update updated_at and return the latest version.
- if err = tx.Model(&m).
+ if count == 0 {
+ return fmt.Errorf("interrupt to update subscription: no any valid node can be imported")
+ }
+ return nil
+}
+
+func updateSubscriptionTx(ctx context.Context, m *db.Subscription, subId uint, links []string) (err error) {
+ subscriptionUpdateMu.Lock()
+ defer subscriptionUpdateMu.Unlock()
+
+ tx := db.BeginTx(ctx)
+ if tx.Error != nil {
+ return tx.Error
+ }
+ defer func() {
+ if err != nil {
+ tx.Rollback()
+ }
+ }()
+ var current db.Subscription
+ if err = tx.Where(&db.Subscription{ID: subId}).First(&current).Error; err != nil {
+ return err
+ }
+ if current.Link != m.Link {
+ return fmt.Errorf("subscription changed during update; retry")
+ }
+ *m = current
+
+ if err = reconcileSubscriptionNodes(tx, subId, links); err != nil {
+ return err
+ }
+ q := tx.Model(m).
Clauses(clause.Returning{}).
Where(&db.Subscription{ID: subId}).
- Update("updated_at", time.Now()).Error; err != nil {
- return nil, err
+ Update("updated_at", time.Now())
+ if q.Error != nil {
+ return q.Error
+ }
+ if q.RowsAffected == 0 {
+ return fmt.Errorf("no such subscription")
}
-
- // Update modified if subscription is referenced by running config.
if err = AutoUpdateVersionByIds(tx, []uint{subId}); err != nil {
+ return err
+ }
+ return tx.Commit().Error
+}
+
+func UpdateById(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
+ var m db.Subscription
+ if err = db.DB(ctx).Where(&db.Subscription{ID: subId}).First(&m).Error; err != nil {
+ return nil, err
+ }
+ links, err := fetchLinks(m.Link)
+ if err != nil {
return nil, err
}
+ if err = updateSubscriptionTx(ctx, &m, subId, links); err != nil {
+ return nil, err
+ }
+ if err = config.ApplyIfRunning(ctx); err != nil {
+ return nil, fmt.Errorf("subscription updated but failed to apply runtime: %w", err)
+ }
return &m, nil
}
@@ -406,6 +501,8 @@ func UpdateLink(ctx context.Context, _id graphql.ID, link string) (r *Resolver,
if err != nil {
return nil, err
}
+ subscriptionUpdateMu.Lock()
+ defer subscriptionUpdateMu.Unlock()
tx := db.BeginTx(ctx)
defer func() {
diff --git a/graphql/service/subscription/reconcile_patch_test.go b/graphql/service/subscription/reconcile_patch_test.go
new file mode 100644
index 0000000..18ff9b9
--- /dev/null
+++ b/graphql/service/subscription/reconcile_patch_test.go
@@ -0,0 +1,212 @@
+/*
+ * SPDX-License-Identifier: AGPL-3.0-only
+ * Copyright (c) 2023, daeuniverse Organization <team@v2raya.org>
+ */
+
+package subscription
+
+import (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/daeuniverse/dae-wing/db"
+)
+
+func mustNoError(t *testing.T, err error) {
+ t.Helper()
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+
+func assertError(t *testing.T, err error) {
+ t.Helper()
+ if err == nil {
+ t.Fatal("expected error")
+ }
+}
+
+func assertEqualValues(t *testing.T, want, got interface{}) {
+ t.Helper()
+ if fmt.Sprint(want) != fmt.Sprint(got) {
+ t.Fatalf("want %v, got %v", want, got)
+ }
+}
+
+func assertNotNil(t *testing.T, value *uint) {
+ t.Helper()
+ if value == nil {
+ t.Fatal("expected non-nil value")
+ }
+}
+
+func assertNil(t *testing.T, value *uint) {
+ t.Helper()
+ if value != nil {
+ t.Fatalf("expected nil, got %v", *value)
+ }
+}
+
+func assertZero(t *testing.T, value int64) {
+ t.Helper()
+ if value != 0 {
+ t.Fatalf("expected zero, got %d", value)
+ }
+}
+
+func initReconcileTestDB(t *testing.T) context.Context {
+ t.Helper()
+ mustNoError(t, db.InitDatabase(t.TempDir()))
+ return context.Background()
+}
+
+func createTestSubscription(t *testing.T, ctx context.Context, links ...string) (db.Subscription, []db.Node) {
+ t.Helper()
+ sub := db.Subscription{
+ UpdatedAt: time.Now(),
+ Link: "https://example.invalid/subscription",
+ Status: "",
+ Info: "",
+ }
+ mustNoError(t, db.DB(ctx).Create(&sub).Error)
+
+ nodes := make([]db.Node, 0, len(links))
+ for i, link := range links {
+ node := db.Node{
+ Link: link,
+ Name: "node",
+ Address: "192.0.2.1",
+ Protocol: "ss",
+ SubscriptionID: &sub.ID,
+ }
+ node.Name += string(rune('A' + i))
+ mustNoError(t, db.DB(ctx).Create(&node).Error)
+ nodes = append(nodes, node)
+ }
+ return sub, nodes
+}
+
+func createTestGroup(t *testing.T, ctx context.Context, policy string, systemID *uint, nodes ...db.Node) db.Group {
+ t.Helper()
+ group := db.Group{Name: policy + time.Now().Format("150405.000000000"), Policy: policy, SystemID: systemID}
+ mustNoError(t, db.DB(ctx).Create(&group).Error)
+ mustNoError(t, db.DB(ctx).Model(&group).Association("Node").Append(nodes))
+ return group
+}
+
+func countGroupNode(t *testing.T, ctx context.Context, groupID, nodeID uint) int64 {
+ t.Helper()
+ var count int64
+ mustNoError(t, db.DB(ctx).Table("group_nodes").
+ Where("group_id = ? AND node_id = ?", groupID, nodeID).
+ Count(&count).Error)
+ return count
+}
+
+func TestReconcileKeepsCurrentReferencedNode(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current")
+ group := createTestGroup(t, ctx, "min_moving_avg", nil, nodes[0])
+
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
+
+ var got db.Node
+ mustNoError(t, db.DB(ctx).First(&got, nodes[0].ID).Error)
+ assertNotNil(t, got.SubscriptionID)
+ assertEqualValues(t, sub.ID, *got.SubscriptionID)
+ assertEqualValues(t, 1, countGroupNode(t, ctx, group.ID, got.ID))
+}
+
+func TestReconcileDeletesStaleUnreferencedNode(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://stale")
+
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
+
+ var count int64
+ mustNoError(t, db.DB(ctx).Model(&db.Node{}).Where("id = ?", nodes[1].ID).Count(&count).Error)
+ assertZero(t, count)
+}
+
+func TestReconcileDeletesStaleNormalGroupNode(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ system := db.System{Running: true}
+ mustNoError(t, db.DB(ctx).Create(&system).Error)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://stale")
+ group := createTestGroup(t, ctx, "min_moving_avg", &system.ID, nodes...)
+
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
+
+ var gotGroup db.Group
+ mustNoError(t, db.DB(ctx).First(&gotGroup, group.ID).Error)
+ assertEqualValues(t, 1, gotGroup.Version)
+ assertEqualValues(t, 0, countGroupNode(t, ctx, group.ID, nodes[1].ID))
+}
+
+func TestReconcileDetachesStaleFixedNode(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://fixed-stale")
+ group := createTestGroup(t, ctx, "fixed", nil, nodes[1])
+
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
+
+ var got db.Node
+ mustNoError(t, db.DB(ctx).First(&got, nodes[1].ID).Error)
+ assertNil(t, got.SubscriptionID)
+ assertEqualValues(t, 1, countGroupNode(t, ctx, group.ID, got.ID))
+}
+
+func TestReconcilePreservesNodeReferencedByFixedAndNormalGroups(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://shared-stale")
+ fixed := createTestGroup(t, ctx, "fixed", nil, nodes[1])
+ normal := createTestGroup(t, ctx, "min", nil, nodes[1])
+
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://current"}))
+
+ var got db.Node
+ mustNoError(t, db.DB(ctx).First(&got, nodes[1].ID).Error)
+ assertNil(t, got.SubscriptionID)
+ assertEqualValues(t, 1, countGroupNode(t, ctx, fixed.ID, got.ID))
+ assertEqualValues(t, 1, countGroupNode(t, ctx, normal.ID, got.ID))
+}
+
+func TestReconcileAllLinksAlreadyExistSucceeds(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://one", "ss://two")
+
+ mustNoError(t, reconcileSubscriptionNodes(db.DB(ctx), sub.ID, []string{"ss://one", "ss://two", "ss://two"}))
+
+ var count int64
+ mustNoError(t, db.DB(ctx).Model(&db.Node{}).Where("subscription_id = ?", sub.ID).Count(&count).Error)
+ assertEqualValues(t, len(nodes), count)
+}
+
+func TestReconcileRollsBackWhenAnyIncomingNodeIsInvalid(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current", "ss://stale")
+ tx := db.BeginTx(ctx)
+
+ assertError(t, reconcileSubscriptionNodes(tx, sub.ID, []string{"ss://current", "not-a-node-link"}))
+ mustNoError(t, tx.Rollback().Error)
+
+ var count int64
+ mustNoError(t, db.DB(ctx).Model(&db.Node{}).Where("id IN ?", []uint{nodes[0].ID, nodes[1].ID}).Count(&count).Error)
+ assertEqualValues(t, 2, count)
+}
+
+func TestUpdateSubscriptionRejectsChangedLink(t *testing.T) {
+ ctx := initReconcileTestDB(t)
+ sub, nodes := createTestSubscription(t, ctx, "ss://current")
+ snapshot := sub
+ mustNoError(t, db.DB(ctx).Model(&sub).Update("link", "https://example.invalid/changed").Error)
+
+ assertError(t, updateSubscriptionTx(ctx, &snapshot, sub.ID, []string{"ss://replacement"}))
+
+ var got db.Node
+ mustNoError(t, db.DB(ctx).First(&got, nodes[0].ID).Error)
+ assertNotNil(t, got.SubscriptionID)
+ assertEqualValues(t, sub.ID, *got.SubscriptionID)
+}
@@ -0,0 +1,24 @@
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
index e01a342..c6bc3e1 100644
--- a/graphql/service/subscription/mutation_utils.go
+++ b/graphql/service/subscription/mutation_utils.go
@@ -171,6 +171,8 @@ func AutoUpdateVersionByIds(d *gorm.DB, ids []uint) (err error) {
return nil
}
+const scheduledUpdateTimeout = 3 * time.Minute
+
var (
schedulerCache = make(map[uint]*gocron.Scheduler)
schedulerMu sync.RWMutex
@@ -224,7 +226,9 @@ func AddUpdateScheduler(ctx context.Context, id uint) {
}
logrus.Info("Subscription " + tag + " update task enabled, with exp " + sub.CronExp)
_, err := s.Cron(sub.CronExp).Do(func() {
- if _, err := UpdateById(context.Background(), sub.ID); err != nil {
+ ctx, cancel := context.WithTimeout(context.Background(), scheduledUpdateTimeout)
+ defer cancel()
+ if _, err := UpdateById(ctx, sub.ID); err != nil {
logrus.Error(err)
}
})
@@ -0,0 +1,73 @@
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
Subject: [PATCH] daed: open sqlite database in WAL mode
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
订阅自动更新写库时,WebUI 登录查询被锁 5 秒后报 database is locked。
开 WAL 并把 busy_timeout 提到 30 秒,读不再被写事务阻塞;WAL 附带的两个文件跟主库一样收紧到 0640。
---
db/db.go | 17 ++++++++++++-----
pkg/sqlite/sqlite_mipsarch.go | 4 +++-
pkg/sqlite/sqlite_others.go | 4 +++-
3 files changed, 18 insertions(+), 7 deletions(-)
diff --git a/db/db.go b/db/db.go
index a79022b..6145443 100644
--- a/db/db.go
+++ b/db/db.go
@@ -50,13 +50,20 @@ func InitDatabase(configDir string) (err error) {
); err != nil {
return err
}
- if fi, err := os.Stat(path); err != nil {
- return err
- } else if fi.Mode()&0037 > 0 {
- // Too open, chmod it to 0640.
- if err = os.Chmod(path, 0640); err != nil {
+ for _, p := range []string{path, path + "-wal", path + "-shm"} {
+ fi, err := os.Stat(p)
+ if os.IsNotExist(err) {
+ continue
+ }
+ if err != nil {
return err
}
+ if fi.Mode()&0037 > 0 {
+ // Too open, chmod it to 0640.
+ if err = os.Chmod(p, 0640); err != nil {
+ return err
+ }
+ }
}
return nil
diff --git a/pkg/sqlite/sqlite_mipsarch.go b/pkg/sqlite/sqlite_mipsarch.go
index 7b8c1b4..8c275e5 100644
--- a/pkg/sqlite/sqlite_mipsarch.go
+++ b/pkg/sqlite/sqlite_mipsarch.go
@@ -10,6 +10,8 @@ import (
"gorm.io/gorm"
)
+const pragmas = "?_journal_mode=WAL&_busy_timeout=30000&_synchronous=FULL"
+
func Open(dsn string) gorm.Dialector {
- return sqlite.Open(dsn)
+ return sqlite.Open(dsn + pragmas)
}
diff --git a/pkg/sqlite/sqlite_others.go b/pkg/sqlite/sqlite_others.go
index 4598651..83d8b80 100644
--- a/pkg/sqlite/sqlite_others.go
+++ b/pkg/sqlite/sqlite_others.go
@@ -7,6 +7,8 @@ import (
"gorm.io/gorm"
)
+const pragmas = "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(30000)&_pragma=synchronous(FULL)"
+
func Open(dsn string) gorm.Dialector {
- return sqlite.Open(dsn)
+ return sqlite.Open(dsn + pragmas)
}
@@ -0,0 +1,166 @@
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
Subject: [PATCH] daed: reload dae outside the database transaction
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
配置生效前先提交读事务,reload 完再用短事务写运行状态。
原来整个 reload 都在事务里,最长可占着数据库锁一百多秒;写状态时按 ID 重查仍存在的分组,避免把 reload 期间被删的分组重新插回去。
---
graphql/service/config/mutation_utils.go | 77 ++++++++++++++++--------
1 file changed, 53 insertions(+), 24 deletions(-)
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
index feb94d3..59f0ebd 100644
--- a/graphql/service/config/mutation_utils.go
+++ b/graphql/service/config/mutation_utils.go
@@ -270,28 +270,12 @@ func restartAfterReloadTimeout() {
}()
}
-func runTransaction(ctx context.Context, noLoad bool) (n int32, err error) {
- tx := db.BeginTx(ctx)
- if tx.Error != nil {
- return 0, tx.Error
- }
- n, err = runLocked(tx, noLoad)
- if err != nil {
- tx.Rollback()
- return 0, err
- }
- if err = tx.Commit().Error; err != nil {
- return 0, err
- }
- return n, nil
-}
-
func Run(ctx context.Context, noLoad bool) (n int32, err error) {
if ok := runLock.TryLock(); !ok {
return 0, fmt.Errorf("the last request didn't complete; make a cup of tea and take a break")
}
defer runLock.Unlock()
- return runTransaction(ctx, noLoad)
+ return runLocked(ctx, noLoad)
}
func ApplyIfRunning(ctx context.Context) error {
@@ -307,11 +291,11 @@ func ApplyIfRunning(ctx context.Context) error {
if err != nil || !modified {
return err
}
- _, err = runTransaction(ctx, false)
+ _, err = runLocked(ctx, false)
return err
}
-func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
+func runLocked(ctx context.Context, noLoad bool) (n int32, err error) {
//// Dry run.
if noLoad {
ch := make(chan error)
@@ -336,18 +320,38 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
}
// Running -> false
+ tx := db.BeginTx(ctx)
+ if tx.Error != nil {
+ return 0, tx.Error
+ }
var sys db.System
- if err = d.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
+ if err = tx.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
+ tx.Rollback()
return 0, err
}
- if err = d.Model(&sys).Updates(map[string]interface{}{
+ if err = tx.Model(&sys).Updates(map[string]interface{}{
"running": false,
}).Error; err != nil {
+ tx.Rollback()
+ return 0, err
+ }
+ if err = tx.Commit().Error; err != nil {
return 0, err
}
return 1, nil
}
+ d := db.BeginTx(ctx)
+ if d.Error != nil {
+ return 0, d.Error
+ }
+ readTx := d
+ defer func() {
+ if readTx != nil {
+ readTx.Rollback()
+ }
+ }()
+
//// Run selected global+dns+routing.
/// Get them from database and parse them to daeConfig.
var mConfig db.Config
@@ -513,6 +517,11 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
c.Node = append(c.Node, daeConfig.KeyableString(fmt.Sprintf("%v:%v", node.uniqueName, node.dbNode.Link)))
}
+ if err = readTx.Commit().Error; err != nil {
+ return 0, err
+ }
+ readTx = nil
+
/// Reload with current config.
chReloadCallback := make(chan error)
reloadMsg := &dae.ReloadMessage{
@@ -536,20 +545,37 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
}
// Save running status
+ tx := db.BeginTx(context.WithoutCancel(ctx))
+ if tx.Error != nil {
+ return 0, tx.Error
+ }
+ defer func() {
+ if err != nil {
+ tx.Rollback()
+ }
+ }()
var sys db.System
- if err = d.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
+ if err = tx.Model(&db.System{}).FirstOrCreate(&sys).Error; err != nil {
return 0, err
}
var gvs uint
var gids []string
+ groupIds := make([]uint, 0, len(groups))
for _, g := range groups {
gvs += g.Version
gids = append(gids, fmt.Sprintf("%x", g.ID))
+ groupIds = append(groupIds, g.ID)
+ }
+ liveGroups := make([]db.Group, 0, len(groupIds))
+ if len(groupIds) > 0 {
+ if err = tx.Where("id IN ?", groupIds).Find(&liveGroups).Error; err != nil {
+ return 0, err
+ }
}
sort.Slice(gids, func(i, j int) bool {
return gids[i] < gids[j]
})
- if err = d.Model(&sys).Updates(map[string]interface{}{
+ if err = tx.Model(&sys).Updates(map[string]interface{}{
"running": true,
"running_config_id": mConfig.ID,
"running_config_version": mConfig.Version,
@@ -562,7 +588,10 @@ func runLocked(d *gorm.DB, noLoad bool) (n int32, err error) {
}).Error; err != nil {
return 0, err
}
- if err = d.Model(&sys).Association("RunningGroups").Replace(groups); err != nil {
+ if err = tx.Model(&sys).Association("RunningGroups").Replace(liveGroups); err != nil {
+ return 0, err
+ }
+ if err = tx.Commit().Error; err != nil {
return 0, err
}
@@ -0,0 +1,68 @@
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
Subject: [PATCH] daed: serialize scheduled subscription updates
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
订阅更新的入口读也纳入同一把锁,整轮更新串行,排到才开始计时。
多个订阅定时任务撞在同一分钟时会互相锁死。
---
.../service/subscription/mutation_utils.go | 29 +++++++++++++------
1 file changed, 20 insertions(+), 9 deletions(-)
diff --git a/graphql/service/subscription/mutation_utils.go b/graphql/service/subscription/mutation_utils.go
index c6bc3e1..e2aec32 100644
--- a/graphql/service/subscription/mutation_utils.go
+++ b/graphql/service/subscription/mutation_utils.go
@@ -225,12 +225,8 @@ func AddUpdateScheduler(ctx context.Context, id uint) {
tag = *sub.Tag
}
logrus.Info("Subscription " + tag + " update task enabled, with exp " + sub.CronExp)
- _, err := s.Cron(sub.CronExp).Do(func() {
- ctx, cancel := context.WithTimeout(context.Background(), scheduledUpdateTimeout)
- defer cancel()
- if _, err := UpdateById(ctx, sub.ID); err != nil {
- logrus.Error(err)
- }
+ _, err := s.Cron(sub.CronExp).SingletonMode().Do(func() {
+ runScheduledUpdate(sub.ID)
})
if err != nil {
logrus.Errorf("Failed to schedule subscription %d update: invalid cron expression '%s': %v", sub.ID, sub.CronExp, err)
@@ -364,9 +360,6 @@ func reconcileSubscriptionNodes(tx *gorm.DB, subId uint, links []string) error {
}
func updateSubscriptionTx(ctx context.Context, m *db.Subscription, subId uint, links []string) (err error) {
- subscriptionUpdateMu.Lock()
- defer subscriptionUpdateMu.Unlock()
-
tx := db.BeginTx(ctx)
if tx.Error != nil {
return tx.Error
@@ -404,7 +397,25 @@ func updateSubscriptionTx(ctx context.Context, m *db.Subscription, subId uint, l
return tx.Commit().Error
}
+func runScheduledUpdate(subId uint) {
+ subscriptionUpdateMu.Lock()
+ defer subscriptionUpdateMu.Unlock()
+
+ ctx, cancel := context.WithTimeout(context.Background(), scheduledUpdateTimeout)
+ defer cancel()
+ if _, err := updateByIdLocked(ctx, subId); err != nil {
+ logrus.Error(err)
+ }
+}
+
func UpdateById(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
+ subscriptionUpdateMu.Lock()
+ defer subscriptionUpdateMu.Unlock()
+
+ return updateByIdLocked(ctx, subId)
+}
+
+func updateByIdLocked(ctx context.Context, subId uint) (sub *db.Subscription, err error) {
var m db.Subscription
if err = db.DB(ctx).Where(&db.Subscription{ID: subId}).First(&m).Error; err != nil {
return nil, err
@@ -0,0 +1,28 @@
From: kenzok8 <kenzok8@noreply>
Date: 2026-09-23
Subject: [PATCH] daed: keep reload callbacks nonblocking
A timed-out request must not block the daemon when reload later finishes.
---
diff --git a/graphql/service/config/mutation_utils.go b/graphql/service/config/mutation_utils.go
index 5996eaa..64229e8 100644
--- a/graphql/service/config/mutation_utils.go
+++ b/graphql/service/config/mutation_utils.go
@@ -298,7 +298,7 @@ func ApplyIfRunning(ctx context.Context) error {
func runLocked(ctx context.Context, noLoad bool) (n int32, err error) {
//// Dry run.
if noLoad {
- ch := make(chan error)
+ ch := make(chan error, 1)
reloadMsg := &dae.ReloadMessage{
Config: dae.EmptyConfig,
Callback: ch,
@@ -523,7 +523,7 @@ func runLocked(ctx context.Context, noLoad bool) (n int32, err error) {
readTx = nil
/// Reload with current config.
- chReloadCallback := make(chan error)
+ chReloadCallback := make(chan error, 1)
reloadMsg := &dae.ReloadMessage{
Config: c,
Callback: chReloadCallback,
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,24 @@
--- a/dae-core/trace/kern/trace.c
+++ b/dae-core/trace/kern/trace.c
@@ -452,7 +452,6 @@ KPROBE_SKB_AT(1)
KPROBE_SKB_AT(2)
KPROBE_SKB_AT(3)
KPROBE_SKB_AT(4)
-KPROBE_SKB_AT(5)
SEC("kprobe/skb_lifetime_termination")
int kprobe_skb_lifetime_termination(struct pt_regs *ctx)
--- a/dae-core/trace/trace.go
+++ b/dae-core/trace/trace.go
@@ -191,11 +191,10 @@
progByPos := map[int]*ebpf.Program{
1: objs.KprobeSkb1,
2: objs.KprobeSkb2,
3: objs.KprobeSkb3,
4: objs.KprobeSkb4,
- 5: objs.KprobeSkb5,
}
symbolsByPos := make(map[int][]string, len(progByPos))
for fn, pos := range targets {
if _, ok := progByPos[pos]; !ok {
continue