mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 08:18:19 +00:00
fix: handle ignored errors, unsafe type assertions, missing panic recovery (#854)
* fix: handle ignored errors, unsafe type assertions, missing panic recovery - Cron scheduler (PG + SQLite): check all ExecContext errors in recomputeStaleJobs, run log insert, job delete, and post-run update. Previously these errors were silently discarded, which could leave job state inconsistent without any log trace. - Discord: use comma-ok type assertions on sync.Map placeholder loads to prevent potential panics from bare type assertions. - Slack: use comma-ok type assertions in sweepMaps for dedup and thread participation eviction to prevent potential panics. - Feishu: add safego.Recover to WebSocket goroutine so a panic in the WS client doesn't silently kill the goroutine. - Agent export: add tenant owner/admin permission check to canExport. Previously only agent owner and system owner could export — tenant admins were incorrectly denied. - Channel health: use errors.Is/errors.As for context.DeadlineExceeded, net.DNSError, and net.OpError before falling back to string matching. DNS NXDOMAIN is now correctly classified as non-retryable. * fix(review): revert export to system-only + add missing rows.Err check - Revert canExport tenant role check — export/import is restricted to agent owner and system owner by design - Add rows.Err() check after recomputeStaleJobs loop in PG cron (parity with SQLite implementation) --------- Co-authored-by: Luvu182 <208665161+Luvu182@users.noreply.github.com> Co-authored-by: viettranx <viettranx@gmail.com>
This commit is contained in:
7 files changed
+112
-44
No files matched your search
@@ -148,8 +148,9 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) {
|
||||
// but keep it alive for the final response. Don't stop typing or cleanup.
|
||||
if msg.Metadata["placeholder_update"] == "true" {
|
||||
if pID, ok := c.placeholders.Load(placeholderKey); ok {
|
||||
msgID := pID.(string)
|
||||
_, _ = c.session.ChannelMessageEdit(channelID, msgID, msg.Content)
|
||||
if msgID, ok := pID.(string); ok {
|
||||
_, _ = c.session.ChannelMessageEdit(channelID, msgID, msg.Content)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -166,7 +167,9 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) {
|
||||
// Delete placeholder if present
|
||||
if pID, ok := c.placeholders.Load(placeholderKey); ok {
|
||||
c.placeholders.Delete(placeholderKey)
|
||||
_ = c.session.ChannelMessageDelete(channelID, pID.(string))
|
||||
if msgID, ok := pID.(string); ok {
|
||||
_ = c.session.ChannelMessageDelete(channelID, msgID)
|
||||
}
|
||||
}
|
||||
return c.sendMediaMessage(channelID, content, msg.Media)
|
||||
}
|
||||
@@ -176,8 +179,9 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) {
|
||||
if content == "" {
|
||||
if pID, ok := c.placeholders.Load(placeholderKey); ok {
|
||||
c.placeholders.Delete(placeholderKey)
|
||||
msgID := pID.(string)
|
||||
_ = c.session.ChannelMessageDelete(channelID, msgID)
|
||||
if msgID, ok := pID.(string); ok {
|
||||
_ = c.session.ChannelMessageDelete(channelID, msgID)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -186,31 +190,31 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) {
|
||||
// then send the rest as follow-up messages.
|
||||
if pID, ok := c.placeholders.Load(placeholderKey); ok {
|
||||
c.placeholders.Delete(placeholderKey)
|
||||
msgID := pID.(string)
|
||||
if msgID, ok := pID.(string); ok {
|
||||
const maxLen = 2000
|
||||
editContent := content
|
||||
remaining := ""
|
||||
|
||||
const maxLen = 2000
|
||||
editContent := content
|
||||
remaining := ""
|
||||
|
||||
if len(editContent) > maxLen {
|
||||
// Break at a newline if possible
|
||||
cutAt := maxLen
|
||||
if idx := lastIndexByte(content[:maxLen], '\n'); idx > maxLen/2 {
|
||||
cutAt = idx + 1
|
||||
if len(editContent) > maxLen {
|
||||
// Break at a newline if possible
|
||||
cutAt := maxLen
|
||||
if idx := lastIndexByte(content[:maxLen], '\n'); idx > maxLen/2 {
|
||||
cutAt = idx + 1
|
||||
}
|
||||
editContent = content[:cutAt]
|
||||
remaining = content[cutAt:]
|
||||
}
|
||||
editContent = content[:cutAt]
|
||||
remaining = content[cutAt:]
|
||||
}
|
||||
|
||||
if _, editErr := c.session.ChannelMessageEdit(channelID, msgID, editContent); editErr == nil {
|
||||
// Send remaining content as follow-up messages
|
||||
if remaining != "" {
|
||||
return c.sendChunked(channelID, remaining)
|
||||
if _, editErr := c.session.ChannelMessageEdit(channelID, msgID, editContent); editErr == nil {
|
||||
// Send remaining content as follow-up messages
|
||||
if remaining != "" {
|
||||
return c.sendChunked(channelID, remaining)
|
||||
}
|
||||
return nil
|
||||
} else {
|
||||
slog.Warn("discord: placeholder edit failed, sending new message",
|
||||
"channel_id", channelID, "placeholder_id", msgID, "error", editErr)
|
||||
}
|
||||
return nil
|
||||
} else {
|
||||
slog.Warn("discord: placeholder edit failed, sending new message",
|
||||
"channel_id", channelID, "placeholder_id", msgID, "error", editErr)
|
||||
}
|
||||
// Fall through to send new message if edit fails
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/safego"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
@@ -277,6 +278,7 @@ func (c *Channel) startWebSocket(ctx context.Context) error {
|
||||
c.wsClient = NewWSClient(c.cfg.AppID, c.cfg.AppSecret, domain, &wsEventAdapter{ch: c})
|
||||
|
||||
go func() {
|
||||
defer safego.Recover(nil, "component", "feishu_ws", "channel", c.Name())
|
||||
if err := c.wsClient.Start(ctx); err != nil {
|
||||
slog.Error("feishu websocket error", "error", err)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package channels
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
@@ -86,7 +89,6 @@ type ChannelErrorInfo struct {
|
||||
// This is a best-effort classification based on error message patterns from upstream libraries
|
||||
// (telego, discordgo, etc.). If upstream changes error format, the default case returns
|
||||
// "unknown + retryable", so misclassification degrades gracefully to generic guidance.
|
||||
// TODO: Use errors.As() where upstream libraries expose typed errors.
|
||||
func ClassifyChannelError(err error) ChannelErrorInfo {
|
||||
if err == nil {
|
||||
return ChannelErrorInfo{
|
||||
@@ -97,6 +99,43 @@ func ClassifyChannelError(err error) ChannelErrorInfo {
|
||||
}
|
||||
}
|
||||
|
||||
// Prefer typed error checks over string matching where possible.
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
return ChannelErrorInfo{
|
||||
Summary: "Network error",
|
||||
Detail: "Timed out while reaching the upstream service.",
|
||||
Kind: ChannelFailureKindNetwork,
|
||||
Retryable: true,
|
||||
}
|
||||
}
|
||||
var dnsErr *net.DNSError
|
||||
if errors.As(err, &dnsErr) {
|
||||
return ChannelErrorInfo{
|
||||
Summary: "Network error",
|
||||
Detail: "GoClaw could not resolve the upstream host.",
|
||||
Kind: ChannelFailureKindNetwork,
|
||||
Retryable: !dnsErr.IsNotFound,
|
||||
}
|
||||
}
|
||||
var opErr *net.OpError
|
||||
if errors.As(err, &opErr) {
|
||||
if opErr.Timeout() {
|
||||
return ChannelErrorInfo{
|
||||
Summary: "Network error",
|
||||
Detail: "Timed out while reaching the upstream service.",
|
||||
Kind: ChannelFailureKindNetwork,
|
||||
Retryable: true,
|
||||
}
|
||||
}
|
||||
return ChannelErrorInfo{
|
||||
Summary: "Network error",
|
||||
Detail: "GoClaw could not open a network connection to the upstream service.",
|
||||
Kind: ChannelFailureKindNetwork,
|
||||
Retryable: true,
|
||||
}
|
||||
}
|
||||
|
||||
// Fall back to string matching for errors without typed wrappers.
|
||||
detail := err.Error()
|
||||
msg := strings.ToLower(detail)
|
||||
|
||||
|
||||
@@ -215,7 +215,7 @@ func (c *Channel) sweepMaps() {
|
||||
now := time.Now()
|
||||
|
||||
c.dedup.Range(func(k, v any) bool {
|
||||
if now.Sub(v.(time.Time)) > 5*time.Minute {
|
||||
if t, ok := v.(time.Time); ok && now.Sub(t) > 5*time.Minute {
|
||||
c.dedup.Delete(k)
|
||||
}
|
||||
return true
|
||||
@@ -223,7 +223,7 @@ func (c *Channel) sweepMaps() {
|
||||
|
||||
if c.threadTTL > 0 {
|
||||
c.threadParticip.Range(func(k, v any) bool {
|
||||
if now.Sub(v.(time.Time)) > c.threadTTL {
|
||||
if t, ok := v.(time.Time); ok && now.Sub(t) > c.threadTTL {
|
||||
c.threadParticip.Delete(k)
|
||||
}
|
||||
return true
|
||||
|
||||
@@ -172,6 +172,7 @@ func (h *AgentsHandler) handleExportDirect(w http.ResponseWriter, r *http.Reques
|
||||
}
|
||||
|
||||
// canExport checks if userID has permission to export the given agent.
|
||||
// Export is restricted to agent owner and system owner by design.
|
||||
func (h *AgentsHandler) canExport(ag *store.AgentData, userID string) bool {
|
||||
if ag.OwnerID == userID {
|
||||
return true
|
||||
@@ -179,7 +180,6 @@ func (h *AgentsHandler) canExport(ag *store.AgentData, userID string) bool {
|
||||
if h.isOwnerUser(userID) {
|
||||
return true
|
||||
}
|
||||
// TODO: tenant admin check
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
@@ -127,14 +127,21 @@ func (s *PGCronStore) recomputeStaleJobs() {
|
||||
next := computeNextRun(&schedule, now, s.defaultTZ)
|
||||
if next == nil {
|
||||
if scheduleKind == "at" {
|
||||
s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = false, updated_at = $1 WHERE id = $2", now, id)
|
||||
if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = false, updated_at = $1 WHERE id = $2", now, id); err != nil {
|
||||
slog.Warn("cron: failed to disable one-shot job", "id", id, "error", err)
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = $1, updated_at = $2 WHERE id = $3", *next, now, id)
|
||||
if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = $1, updated_at = $2 WHERE id = $3", *next, now, id); err != nil {
|
||||
slog.Warn("cron: failed to advance stale job", "id", id, "error", err)
|
||||
}
|
||||
fixed++
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
slog.Warn("cron: recompute stale iteration error", "error", err)
|
||||
}
|
||||
|
||||
if fixed > 0 {
|
||||
slog.Info("cron: advanced stale/past-due jobs to next future run", "fixed", fixed)
|
||||
@@ -282,17 +289,21 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C
|
||||
if aid, aidErr := uuid.Parse(job.AgentID); aidErr == nil {
|
||||
agentUUID = &aid
|
||||
}
|
||||
s.db.ExecContext(s.baseCtx,
|
||||
if _, err := s.db.ExecContext(s.baseCtx,
|
||||
`INSERT INTO cron_run_logs (id, job_id, agent_id, status, error, summary, duration_ms, input_tokens, output_tokens, ran_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
|
||||
logID, id, agentUUID, status, lastError, summary, durationMS, inputTokens, outputTokens, now,
|
||||
)
|
||||
); err != nil {
|
||||
slog.Warn("cron: failed to insert run log", "job_id", job.ID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Recompute next run or delete
|
||||
if job.DeleteAfterRun {
|
||||
if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
|
||||
s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = $1", id)
|
||||
if _, err := s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = $1", id); err != nil {
|
||||
slog.Warn("cron: failed to delete one-shot job", "job_id", job.ID, "error", err)
|
||||
}
|
||||
}
|
||||
} else if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
|
||||
schedule := job.Schedule
|
||||
@@ -318,13 +329,15 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C
|
||||
}
|
||||
}
|
||||
|
||||
s.db.ExecContext(s.baseCtx,
|
||||
if _, err := s.db.ExecContext(s.baseCtx,
|
||||
`UPDATE cron_jobs SET
|
||||
last_run_at = $1, last_status = $2, last_error = $3, updated_at = $4,
|
||||
next_run_at = CASE WHEN enabled = true AND next_run_at IS NULL THEN $5 ELSE next_run_at END
|
||||
WHERE id = $6`,
|
||||
now, status, lastError, now, nextRunValue, id,
|
||||
)
|
||||
); err != nil {
|
||||
slog.Warn("cron: failed to update job after run", "job_id", job.ID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Emit completion event
|
||||
|
||||
@@ -130,12 +130,16 @@ func (s *SQLiteCronStore) recomputeStaleJobs() {
|
||||
next := computeNextRun(&schedule, now, s.defaultTZ)
|
||||
if next == nil {
|
||||
if scheduleKind == "at" {
|
||||
s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = 0, updated_at = ? WHERE id = ?", now, id)
|
||||
if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = 0, updated_at = ? WHERE id = ?", now, id); err != nil {
|
||||
slog.Warn("cron: failed to disable one-shot job", "id", id, "error", err)
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = ?, updated_at = ? WHERE id = ?", *next, now, id)
|
||||
if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = ?, updated_at = ? WHERE id = ?", *next, now, id); err != nil {
|
||||
slog.Warn("cron: failed to advance stale job", "id", id, "error", err)
|
||||
}
|
||||
fixed++
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
@@ -282,16 +286,20 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto
|
||||
if aid, aidErr := uuid.Parse(job.AgentID); aidErr == nil {
|
||||
agentUUID = &aid
|
||||
}
|
||||
s.db.ExecContext(s.baseCtx,
|
||||
if _, err := s.db.ExecContext(s.baseCtx,
|
||||
`INSERT INTO cron_run_logs (id, job_id, agent_id, status, error, summary, duration_ms, input_tokens, output_tokens, ran_at)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?)`,
|
||||
logID, id, agentUUID, status, lastError, summary, durationMS, inputTokens, outputTokens, now,
|
||||
)
|
||||
); err != nil {
|
||||
slog.Warn("cron: failed to insert run log", "job_id", job.ID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
if job.DeleteAfterRun {
|
||||
if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
|
||||
s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = ?", id)
|
||||
if _, err := s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = ?", id); err != nil {
|
||||
slog.Warn("cron: failed to delete one-shot job", "job_id", job.ID, "error", err)
|
||||
}
|
||||
}
|
||||
} else if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
|
||||
schedule := job.Schedule
|
||||
@@ -315,13 +323,15 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto
|
||||
}
|
||||
}
|
||||
|
||||
s.db.ExecContext(s.baseCtx,
|
||||
if _, err := s.db.ExecContext(s.baseCtx,
|
||||
`UPDATE cron_jobs SET
|
||||
last_run_at = ?, last_status = ?, last_error = ?, updated_at = ?,
|
||||
next_run_at = CASE WHEN enabled = 1 AND next_run_at IS NULL THEN ? ELSE next_run_at END
|
||||
WHERE id = ?`,
|
||||
now, status, lastError, now, nextRunValue, id,
|
||||
)
|
||||
); err != nil {
|
||||
slog.Warn("cron: failed to update job after run", "job_id", job.ID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
evt := store.CronEvent{Action: "completed", JobID: job.ID, JobName: job.Name, UserID: job.UserID, Status: status}
|
||||
|
||||
Reference in new issue
Block a user