rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

H-47: bound deferred contact metadata retries (#14)

Browse Source 
Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
2026-08-13 20:22:23 +08:00
committed by GitHub
co-authored by rogee
parent de285b4f5d
commit e33e72877e
3 changed files with 126 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
channels/shangwutong/internal/delivery/inbound.go
+21 -2
View File
@@ -18,7 +18,10 @@ import (
"github.com/sirupsen/logrus"
)
const echoMatchWindow = 10 * time.Minute
const (
echoMatchWindow = 10 * time.Minute
contactMetadataWaitTimeout = 24 * time.Hour
)
type InboundClient interface {
EnsureContact(context.Context, string, string, gochat.ContactRequest) (gochat.Contact, error)
@@ -475,16 +478,27 @@ func (i *Inbound) complete(event *dbgen.InboundEvent, mapped mappedEvent, messag
func (i *Inbound) retryOrFail(event *dbgen.InboundEvent, mapped mappedEvent, deliveryErr error) error {
var apiErr *gochat.APIError
waitingForContact := mapped.ContactCID != "" && isNotFound(deliveryErr)
waitingForContact := mapped.ContactCID != "" && isContactSourceNotFound(deliveryErr)
contactWaitDeadline := event.CreatedAt.Add(contactMetadataWaitTimeout)
if waitingForContact && !time.Now().Before(contactWaitDeadline) {
i.metrics.ContactMetadataWait("expired")
return i.fail(event, mapped, fmt.Errorf("contact source wait expired after %s: %w", contactMetadataWaitTimeout, deliveryErr))
}
retryable := !errors.As(deliveryErr, &apiErr) || apiErr.Retryable || waitingForContact
if retryable && (event.Attempts < 10 || waitingForContact) {
detail := deliveryErr.Error()
next := time.Now().Add(backoffForError(event.Attempts, 5*time.Minute, deliveryErr))
if waitingForContact && next.After(contactWaitDeadline) {
next = contactWaitDeadline
}
persistErr := i.store.Writer().RetryInboundEvent(context.Background(), dbgen.RetryInboundEventParams{
NextAttemptAt: &next, LastError: &detail, ID: event.ID,
})
i.metrics.Mapping(event.Kind, mapped.Strategy, "retry")
i.metrics.Delivery("inbound", "retry")
if waitingForContact {
i.metrics.ContactMetadataWait("waiting")
}
return errors.Join(deliveryErr, persistErr)
}
return i.fail(event, mapped, deliveryErr)
@@ -528,6 +542,11 @@ func isNotFound(err error) bool {
return errors.As(err, &apiErr) && apiErr.StatusCode == 404
}
func isContactSourceNotFound(err error) bool {
var apiErr *gochat.APIError
return errors.As(err, &apiErr) && apiErr.StatusCode == 404 && apiErr.Code == "not_found" && apiErr.Message == "contact source not found"
}
func contentFingerprint(content string) string {
digest := sha256.Sum256([]byte(strings.TrimSpace(content)))
return hex.EncodeToString(digest[:])
channels/shangwutong/internal/delivery/inbound_test.go
+101
View File
@@ -4,11 +4,14 @@ import (
"context"
"errors"
"fmt"
"strings"
"sync"
"testing"
"time"
dbgen "github.com/gochat/gochat/channels/shangwutong/db/generated"
"github.com/gochat/gochat/channels/shangwutong/internal/gochat"
"github.com/gochat/gochat/channels/shangwutong/internal/observability"
"github.com/gochat/gochat/channels/shangwutong/internal/store"
"github.com/gochat/gochat/channels/shangwutong/internal/swt"
)
@@ -102,6 +105,8 @@ func TestInboundKind24WaitsForLaterContactInboxAndConverges(t *testing.T) {
})
client := &inboundRecorder{metadataErr: &gochat.APIError{StatusCode: 404, Code: "not_found", Message: "contact source not found"}}
worker, _ := NewInbound(database, client, nil, 1)
metrics := observability.NewMetrics()
worker.SetMetrics(metrics)
if worked, err := worker.processInbound(ctx); !worked || err == nil {
t.Fatalf("process CID before contact = %v, %v", worked, err)
@@ -110,6 +115,9 @@ func TestInboundKind24WaitsForLaterContactInboxAndConverges(t *testing.T) {
if err != nil || cidEvent.DeliveryStatus != "pending" || cidEvent.NextAttemptAt == nil || cidEvent.LastError == nil {
t.Fatalf("deferred CID event = %#v, %v", cidEvent, err)
}
if payload := string(metrics.Render(observability.MetricSnapshot{})); !strings.Contains(payload, `swt_connector_contact_metadata_wait_total{result="waiting"} 1`) {
t.Fatalf("waiting metric missing from:\n%s", payload)
}
persistInboundEvent(t, database, account, swt.HeartbeatEvent{
SessionID: "visitor", Kind: 2, Text: "hello", SeqID: 43,
@@ -130,6 +138,90 @@ func TestInboundKind24WaitsForLaterContactInboxAndConverges(t *testing.T) {
}
}
func TestInboundKind24PermanentNotFoundFailsImmediately(t *testing.T) {
ctx := context.Background()
database, account := deliveryDatabase(t, ctx)
persistInboundEvent(t, database, account, swt.HeartbeatEvent{
SessionID: "visitor", Kind: 24, Text: "cookie/123", SeqID: 42,
Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 24 [REDACTED] 42 timestamp",
})
client := &inboundRecorder{metadataErr: &gochat.APIError{StatusCode: 404, Code: "not_found", Message: "inbox not found"}}
worker, _ := NewInbound(database, client, nil, 1)
if worked, err := worker.processInbound(ctx); !worked || err == nil {
t.Fatalf("process permanent 404 = %v, %v", worked, err)
}
event, err := database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:24:42")
if err != nil || event.DeliveryStatus != "failed" || event.Attempts != 1 || event.NextAttemptAt != nil {
t.Fatalf("permanent 404 event = %#v, %v", event, err)
}
}
func TestInboundKind24ContactWaitExpiresToDeadLetter(t *testing.T) {
ctx := context.Background()
database, account := deliveryDatabase(t, ctx)
persistInboundEvent(t, database, account, swt.HeartbeatEvent{
SessionID: "visitor", Kind: 24, Text: "cookie/123", SeqID: 42,
Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 24 [REDACTED] 42 timestamp",
})
event, err := database.Writer().ClaimInboundEvent(ctx)
if err != nil {
t.Fatal(err)
}
event.CreatedAt = time.Now().Add(-contactMetadataWaitTimeout - time.Second)
metrics := observability.NewMetrics()
worker, _ := NewInbound(database, &inboundRecorder{}, nil, 1)
worker.SetMetrics(metrics)
deliveryErr := &gochat.APIError{StatusCode: 404, Code: "not_found", Message: "contact source not found"}
if err := worker.retryOrFail(event, mappedEvent{Strategy: "contact_attributes", ContactCID: "cookie/123"}, deliveryErr); err == nil {
t.Fatal("expired contact wait returned nil")
}
stored, err := database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:24:42")
if err != nil || stored.DeliveryStatus != "failed" || stored.LastError == nil || !strings.Contains(*stored.LastError, "contact source wait expired") {
t.Fatalf("expired contact event = %#v, %v", stored, err)
}
payload := string(metrics.Render(observability.MetricSnapshot{}))
if !strings.Contains(payload, `swt_connector_contact_metadata_wait_total{result="expired"} 1`) {
t.Fatalf("expired wait metric missing from:\n%s", payload)
}
}
func TestInboundMultipleWorkersDrainMixedQueueAroundWaitingCID(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
database, account := deliveryDatabase(t, ctx)
for _, event := range []swt.HeartbeatEvent{
{SessionID: "visitor", Kind: 24, Text: "cookie/123", SeqID: 42, Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 24 [REDACTED] 42 timestamp"},
{SessionID: "visitor", Kind: 2, Text: "hello", SeqID: 43, Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 2 hello 43 timestamp"},
{SessionID: "visitor", Kind: 52, Text: "a|cid%2F456|x|y|z|name", SeqID: 44, Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 52 history 44 timestamp"},
} {
persistInboundEvent(t, database, account, event)
}
client := &inboundRecorder{metadataErr: &gochat.APIError{StatusCode: 404, Code: "not_found", Message: "contact source not found"}}
worker, _ := NewInbound(database, client, nil, 1)
if worked, err := worker.processInbound(ctx); !worked || err == nil {
t.Fatalf("prime waiting CID = %v, %v", worked, err)
}
worker, _ = NewInbound(database, client, nil, 4)
worker.Start(ctx)
deadline := time.Now().Add(3 * time.Second)
for {
message, _ := database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:2:43")
cid, _ := database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:24:42")
laterCID, _ := database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:52:44")
if message != nil && cid != nil && laterCID != nil && message.DeliveryStatus == "delivered" && cid.DeliveryStatus == "delivered" && laterCID.DeliveryStatus == "delivered" {
break
}
if time.Now().After(deadline) {
t.Fatalf("mixed queue did not settle: message=%#v cid=%#v laterCID=%#v", message, cid, laterCID)
}
time.Sleep(10 * time.Millisecond)
}
cancel()
worker.Wait()
}
func TestInboundKind3ConfirmsUniqueOutboundEcho(t *testing.T) {
ctx := context.Background()
database, account := deliveryDatabase(t, ctx)
@@ -486,6 +578,7 @@ func completeOutboundParts(t *testing.T, database *store.Store, outboundMessageI
}
type inboundRecorder struct {
mu sync.Mutex
imports []gochat.MessageImport
importErr error
retractions []int64
@@ -498,6 +591,8 @@ type inboundRecorder struct {
}
func (r *inboundRecorder) EnsureContact(_ context.Context, _, _ string, request gochat.ContactRequest) (gochat.Contact, error) {
r.mu.Lock()
defer r.mu.Unlock()
r.metadataErr = nil
return gochat.Contact{ID: 456, SourceID: request.SourceID, Name: request.Name}, nil
}
@@ -507,6 +602,8 @@ func (r *inboundRecorder) UpdateContact(_ context.Context, _, _, sourceID string
}
func (r *inboundRecorder) UpdateContactChannelMetadata(_ context.Context, inboxID int64, sourceID, cid string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.metadataUpdates++
r.lastInboxID, r.lastSourceID, r.lastCID = inboxID, sourceID, cid
return r.metadataErr
@@ -517,6 +614,8 @@ func (r *inboundRecorder) EnsureConversation(context.Context, string, string, ma
}
func (r *inboundRecorder) ImportMessage(_ context.Context, _, _ int64, message gochat.MessageImport) (gochat.ImportedMessage, error) {
r.mu.Lock()
defer r.mu.Unlock()
r.imports = append(r.imports, message)
if r.importErr != nil {
return gochat.ImportedMessage{}, r.importErr
@@ -542,6 +641,8 @@ func (*inboundRecorder) SetVisitorTyping(context.Context, string, string, int64,
}
func (r *inboundRecorder) RetractMessage(_ context.Context, _, _, messageID int64, _ string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.retractions = append(r.retractions, messageID)
return nil
}
channels/shangwutong/internal/observability/metrics.go
+4
View File
@@ -71,6 +71,10 @@ func (m *Metrics) Mapping(kind int64, strategy, result string) {
))
}
func (m *Metrics) ContactMetadataWait(result string) {
m.inc("swt_connector_contact_metadata_wait_total", labels("result", bounded(result, "waiting", "expired")))
}
func (m *Metrics) Unknown(kind int64) {
m.inc("swt_connector_unknown_event_total", labels("kind_group", unknownKindGroup(kind)))
}