Files
dingtalk-workspace-cli/internal/event/bus/lockfile.go
aa6abc5ed6 feat(event): add personal event subscriptions (#589)
* dws event

* fix consume fail

* test: add stream ticket injection probe

* feat: add portal ticket stream mode

* user event

* fix: allow portal ticket normal without app secret

* event

* user event

* eventType filter

* refactor(event): 优化IPC端点路径处理和改进相关测试

- 用dwsevent.IPCEndpoint替代原先根据GOOS判断的路径逻辑
- 新增event包实现Unix socket路径长度限制及长路径fallback机制
- 添加endpoint_test.go覆盖路径短长及唯一性的单元测试
- 修改busctl模块使用统一的IPC端点获取方法,避免重复实现
- transport_unix.go新增checkSocketPath函数检查路径长度,防止EINVAL错误
- 在监听和连接Unix socket时加入路径限制检查,提升错误明晰度
- 去除多个文件中无用的runtime导入,简化代码依赖

* opt

* event skill

* default value

* install script event

* fix: remove subscribe id event fanout filter

* fix(personal): 修正指定发送人消息描述错误

* more im event

* filter subId

* fix: align personal event schema with stream payload

* fix: avoid duplicate app helper name

* feat: simplify personal event schemas

* feat: simplify event schema output

* docs: refine dingtalk event skill references

* feat: align personal event consume flags

* fix event stop and status visibility

* hide app event public entrypoints

* hide incomplete personal sender event

* remove external event reference comments

* chore(event): prepare official release

* fix(event): harden personal stream lifecycle

---------

Co-authored-by: 玉澜 <yulan.wqy@alibaba-inc.com>
Co-authored-by: zhengyubai <zhengyubai618@gmail.com>
2026-07-10 17:54:43 +08:00

158 lines
4.9 KiB
Go

// 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 bus
import (
"errors"
"fmt"
"io"
"os"
"strconv"
"strings"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/lock"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/process"
)
// LockFileName is the on-disk name of the bus single-instance lock. It lives
// inside the bus working directory
// (<ConfigDir>/events/<edition>/<source_kind>/<identity_hash>/).
const LockFileName = "bus.lock"
// ErrBusy is re-exported from lock for callers that only depend on bus.
var ErrBusy = lock.ErrBusy
// ErrStaleOwnerAlive indicates the PID stored in bus.lock points at a
// live process — there is already a bus running for this ClientID and we
// must not start another one. (This case is hit when the holder is still
// alive but its flock was somehow released; in practice flock + PID always
// agree, so this is mostly defensive.)
var ErrStaleOwnerAlive = errors.New("bus: lock file PID is alive but flock was released; assuming live owner")
// Lock represents a held bus.lock. Close releases the flock and removes the
// PID file, so a subsequent bus can acquire cleanly. A zero Lock is unusable.
type Lock struct {
inner *lock.File
}
// Acquire takes the bus lock at path and writes our PID into the file body.
//
// If the file already has a PID written by a previous run:
// 1. Try the flock first — if another process holds it, return ErrBusy
// (a live bus is running, abort).
// 2. flock acquired but file contains a PID → check if that PID is
// alive via process.Alive(). If alive → return ErrStaleOwnerAlive
// (defensive; release our flock first). If dead → take over (orphan
// cleanup) and overwrite PID with our own.
//
// On success the returned Lock owns an exclusive flock and a file body
// containing our PID. Concurrent competing processes will get ErrBusy.
func Acquire(path string) (*Lock, error) {
l, err := lock.TryAcquire(path)
if err != nil {
return nil, err // already wraps ErrBusy when busy
}
// flock acquired. Read existing PID (if any).
f := l.File()
if _, err := f.Seek(0, io.SeekStart); err != nil {
_ = l.Close()
return nil, fmt.Errorf("bus: seek lock: %w", err)
}
old, err := io.ReadAll(f)
if err != nil {
_ = l.Close()
return nil, fmt.Errorf("bus: read lock: %w", err)
}
if pid := parsePID(old); pid > 0 && process.Alive(pid) {
// Defensive: flock returned us the lock, but the stored PID is
// alive. This shouldn't normally happen (the live process holds
// the flock), but it's possible across odd kernel/FS edge cases
// (NFS, container restarts). Release and refuse to start.
_ = l.Close()
return nil, ErrStaleOwnerAlive
}
// Orphan or first-ever acquisition. Rewrite the file with our PID.
if err := truncateAndWritePID(f, os.Getpid()); err != nil {
_ = l.Close()
return nil, fmt.Errorf("bus: write PID: %w", err)
}
return &Lock{inner: l}, nil
}
// ReadHolderPID returns the PID stored in path, or 0 if the file is missing
// or unreadable. Does NOT attempt to acquire the lock — useful for `event
// status` to display the holder without contention.
func ReadHolderPID(path string) int {
b, err := os.ReadFile(path)
if err != nil {
return 0
}
return parsePID(b)
}
// Close releases the flock and best-effort blanks the PID body so a stale
// reader (e.g. `event status` racing our shutdown) does not see our
// long-dead PID and try to signal it. The lock file itself is NOT removed
// — keeping it on disk avoids a race where a competing bus could acquire
// inode-on-create faster than our truncate.
func (l *Lock) Close() error {
if l == nil || l.inner == nil {
return nil
}
// Blank the body before releasing the lock.
f := l.inner.File()
_ = truncateAndWritePID(f, 0)
err := l.inner.Close()
l.inner = nil
return err
}
// HoldsPath returns the path the lock is held on.
func (l *Lock) HoldsPath() string {
if l == nil || l.inner == nil {
return ""
}
return l.inner.Path()
}
func truncateAndWritePID(f *os.File, pid int) error {
if err := f.Truncate(0); err != nil {
return err
}
if _, err := f.Seek(0, io.SeekStart); err != nil {
return err
}
if pid > 0 {
if _, err := fmt.Fprintf(f, "%d\n", pid); err != nil {
return err
}
}
return f.Sync()
}
func parsePID(b []byte) int {
s := strings.TrimSpace(string(b))
if s == "" {
return 0
}
n, err := strconv.Atoi(s)
if err != nil || n <= 0 {
return 0
}
return n
}