Merge pull request #24 from sczheng189/feat/dws-event

fix(event): unix socket 路径超长时 fallback 到短路径，修复深层配置目录下 bus 无法启动
2026-07-02 16:54:58 +08:00
parent 78867f3601 53ce0a8303
commit 836670ef50
6 changed files with 149 additions and 22 deletions
@@ -23,7 +23,6 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
-	"runtime"
 	"sort"
 	"strings"
 	"syscall"
@@ -982,13 +981,10 @@ func editionNameOrDefault() string {
 }

 // defaultIPCEndpoint returns the canonical Unix socket path / Windows pipe
-// name for the bus. Encapsulates the platform-specific shape so callers
-// don't sprinkle GOOS checks throughout the cobra layer.
+// name for the bus. Thin wrapper over dwsevent.IPCEndpoint, which owns the
+// platform shape and the too-long-socket-path fallback.
 func defaultIPCEndpoint(workDir, editionName string, sourceKind dwsevent.SourceKind, identityHash string) string {
-	if runtime.GOOS == "windows" {
-		return `\\.\pipe\dws-event-` + editionName + "-" + sourceKindLabel(sourceKind) + "-" + identityHash
-	}
-	return filepath.Join(workDir, "bus.sock")
+	return dwsevent.IPCEndpoint(workDir, editionName, sourceKind, identityHash)
 }

 func eventWorkDir(configDir, editionName string, sourceKind dwsevent.SourceKind, identityHash string) string {
@@ -30,8 +30,9 @@ import (
 type SpawnFunc func(SpawnConfig) (pid int, err error)

 // DiscoverConfig describes one discover attempt. WorkDir holds bus.lock and
-// (on Unix) bus.sock; the caller must mkdir it with pkg/config.DirPerm
-// beforehand.
+// usually (on Unix) bus.sock — see dwsevent.IPCEndpoint for the short-path
+// fallback when WorkDir is too deep; the caller must mkdir it with
+// pkg/config.DirPerm beforehand.
 type DiscoverConfig struct {
 	WorkDir     string
 	IPCEndpoint string
@@ -17,7 +17,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"runtime"
 	"sort"
 	"time"

@@ -60,23 +59,16 @@ type BusEntry struct {
 	Meta *bus.Meta `json:"meta,omitempty"`
 }

-// IPCEndpoint returns the IPC endpoint for this entry. Unix uses the
-// bus.sock path inside WorkDir; Windows uses the deterministic pipe name
-// derived from edition + clientIDHash (same scheme defaultIPCEndpoint
-// uses in the cobra layer).
+// IPCEndpoint returns the IPC endpoint for this entry. Delegates to
+// dwsevent.IPCEndpoint so status/stop dial exactly where consume and the
+// bus daemon bound (including the short-path fallback when WorkDir is too
+// deep for sun_path).
 func (e BusEntry) IPCEndpoint() string {
 	hash := e.ClientIDHash
 	if e.IdentityHash != "" {
 		hash = e.IdentityHash
 	}
-	if runtime.GOOS == "windows" {
-		kind := string(e.SourceKind)
-		if kind == "" {
-			kind = string(dwsevent.SourceKindAppStream)
-		}
-		return `\\.\pipe\dws-event-` + e.Edition + "-" + kind + "-" + hash
-	}
-	return filepath.Join(e.WorkDir, "bus.sock")
+	return dwsevent.IPCEndpoint(e.WorkDir, e.Edition, e.SourceKind, hash)
 }

 // EnumerateBuses scans <configDir>/events/<editionFilter>/*/ for bus
@@ -0,0 +1,60 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package event
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+)
+
+// MaxUnixSocketPath returns the longest Unix socket path accepted by
+// bind/connect on this OS (Go rejects longer names with EINVAL before
+// the syscall). sockaddr_un.sun_path is 104 bytes on darwin and the
+// BSDs and 108 on Linux; the usable budget is one less.
+func MaxUnixSocketPath() int {
+	if runtime.GOOS == "linux" {
+		return 107
+	}
+	return 103
+}
+
+// IPCEndpoint returns the bus IPC endpoint for one identity: a Named Pipe
+// name on Windows, otherwise bus.sock inside workDir.
+//
+// The canonical Unix location is <workDir>/bus.sock, but workDir derives
+// from the config dir, which can be arbitrarily deep (e.g. dwssb sandboxes
+// use ~/.dwssb/sandboxes/<name>/config/...). When the canonical path would
+// exceed the OS sun_path limit, the socket falls back to a short
+// deterministic path under os.TempDir keyed by a hash of workDir, so every
+// process (consume parent, forked _bus child, status/stop tooling) that
+// derives the endpoint from the same workDir agrees on the location.
+// bus.lock / bus.meta / bus.log always stay in workDir — only the socket
+// moves.
+//
+// This is the single source of truth for endpoint derivation; the cobra
+// layer and busctl must not re-implement the shape.
+func IPCEndpoint(workDir, editionName string, sourceKind SourceKind, identityHash string) string {
+	if sourceKind == "" {
+		sourceKind = SourceKindAppStream
+	}
+	if runtime.GOOS == "windows" {
+		return `\\.\pipe\dws-event-` + editionName + "-" + string(sourceKind) + "-" + identityHash
+	}
+	sock := filepath.Join(workDir, "bus.sock")
+	if len(sock) <= MaxUnixSocketPath() {
+		return sock
+	}
+	return filepath.Join(os.TempDir(), "dws-evt-"+IdentityHash(workDir)+".sock")
+}
@@ -0,0 +1,61 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !windows
+
+package event
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestIPCEndpointShortWorkDirUsesCanonicalPath(t *testing.T) {
+	workDir := "/tmp/dws/events/open/app_stream/aabbccdd00112233"
+	got := IPCEndpoint(workDir, "open", SourceKindAppStream, "aabbccdd00112233")
+	want := filepath.Join(workDir, "bus.sock")
+	if got != want {
+		t.Fatalf("IPCEndpoint = %q, want %q", got, want)
+	}
+}
+
+func TestIPCEndpointLongWorkDirFallsBackUnderTempDir(t *testing.T) {
+	// Mirrors the dwssb sandbox layout that produced a 111-byte socket
+	// path — over macOS's 103-byte usable sun_path budget.
+	workDir := "/Users/zhengyubai/.dwssb/sandboxes/event-subscribe/config/events/open/personal_stream/3928ce0fb4860a52"
+	got := IPCEndpoint(workDir, "open", SourceKindPersonalStream, "3928ce0fb4860a52")
+	if strings.HasPrefix(got, workDir) {
+		t.Fatalf("IPCEndpoint = %q, want fallback outside workDir", got)
+	}
+	if !strings.HasPrefix(got, os.TempDir()) {
+		t.Fatalf("IPCEndpoint = %q, want fallback under os.TempDir %q", got, os.TempDir())
+	}
+	if len(got) > MaxUnixSocketPath() {
+		t.Fatalf("fallback path still too long: %d > %d (%q)", len(got), MaxUnixSocketPath(), got)
+	}
+}
+
+func TestIPCEndpointFallbackIsDeterministicPerWorkDir(t *testing.T) {
+	long := strings.Repeat("x", 120)
+	a := IPCEndpoint("/base/"+long+"/one", "open", SourceKindPersonalStream, "hash")
+	b := IPCEndpoint("/base/"+long+"/one", "open", SourceKindPersonalStream, "hash")
+	c := IPCEndpoint("/base/"+long+"/two", "open", SourceKindPersonalStream, "hash")
+	if a != b {
+		t.Fatalf("same workDir produced different endpoints: %q vs %q", a, b)
+	}
+	if a == c {
+		t.Fatalf("different workDirs collided on endpoint %q", a)
+	}
+}
@@ -20,6 +20,7 @@ import (
 	"net"
 	"os"

+	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
 )

@@ -38,7 +39,20 @@ func (u *unixListener) Close() error {
 	return err
 }

+// checkSocketPath rejects paths over the sun_path budget up front so the
+// caller sees the actual problem instead of the syscall's bare EINVAL
+// ("invalid argument").
+func checkSocketPath(path string) error {
+	if max := dwsevent.MaxUnixSocketPath(); len(path) > max {
+		return fmt.Errorf("transport: unix socket path too long (%d > %d bytes): %s", len(path), max, path)
+	}
+	return nil
+}
+
 func listen(path string) (Listener, error) {
+	if err := checkSocketPath(path); err != nil {
+		return nil, err
+	}
 	// Stale socket cleanup. Caller holds bus.lock so this is race-safe.
 	if _, err := os.Stat(path); err == nil {
 		if err := os.Remove(path); err != nil {
@@ -58,5 +72,8 @@ func listen(path string) (Listener, error) {
 }

 func dial(path string) (net.Conn, error) {
+	if err := checkSocketPath(path); err != nil {
+		return nil, err
+	}
 	return net.Dial("unix", path)
 }