fix kafka message persistence completion

server/im_http_test.go

@@ -41,7 +41,8 @@ type testVerificationStore struct {
 }
 
 type testHTTPInfrastructureStore struct {
-	deadLetters []types.KafkaDeadLetter
+	deadLetters      []types.KafkaDeadLetter
+	markSucceededErr error
 }
 
 type testTenantEntryStore struct {
@@ -110,6 +111,9 @@ func (s *testHTTPInfrastructureStore) BeginKafkaConsume(_ types.TenantID, record
 }
 
 func (s *testHTTPInfrastructureStore) MarkKafkaConsumeSucceeded(_ types.TenantID, _, _ string, _ time.Time) error {
+	if s.markSucceededErr != nil {
+		return s.markSucceededErr
+	}
 	return nil
 }
 

server/im_security_policy_runtime.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/tinode/chat/server/logs"
@@ -20,6 +21,9 @@ const (
 
 	imYouthActionMessage = "message"
 	imYouthActionCall    = "call"
+
+	imSecurityPolicyCacheTTL      = 5 * time.Second
+	imSecurityPolicyErrorCacheTTL = 2 * time.Second
 )
 
 type imPolicyDecision struct {
@@ -29,6 +33,17 @@ type imPolicyDecision struct {
 	Details map[string]any
 }
 
+type imSecurityPolicyCacheEntry struct {
+	policy  *types.SecurityPolicy
+	err     error
+	expires time.Time
+}
+
+var imSecurityPolicyCache = struct {
+	sync.Mutex
+	entries map[string]imSecurityPolicyCacheEntry
+}{entries: make(map[string]imSecurityPolicyCacheEntry)}
+
 func imAllowDecision() imPolicyDecision {
 	return imPolicyDecision{Allowed: true}
 }
@@ -119,7 +134,28 @@ func imSecurityPolicy(tenantID types.TenantID, tenantCode string) (*types.Securi
 	if tenantCode == "" {
 		return nil, types.ErrNotFound
 	}
+	key := fmt.Sprintf("%d:%s", tenantID, tenantCode)
+	now := time.Now()
+	imSecurityPolicyCache.Lock()
+	if cached, ok := imSecurityPolicyCache.entries[key]; ok && now.Before(cached.expires) {
+		imSecurityPolicyCache.Unlock()
+		return cached.policy, cached.err
+	}
+	if store.ModerationConfig == nil {
+		imSecurityPolicyCache.Unlock()
+		return nil, types.ErrUnsupported
+	}
 	policy, err := store.ModerationConfig.SecurityPolicy(tenantCode)
+	ttl := imSecurityPolicyCacheTTL
+	if err != nil {
+		ttl = imSecurityPolicyErrorCacheTTL
+	}
+	imSecurityPolicyCache.entries[key] = imSecurityPolicyCacheEntry{
+		policy:  policy,
+		err:     err,
+		expires: now.Add(ttl),
+	}
+	imSecurityPolicyCache.Unlock()
 	if err != nil {
 		return nil, err
 	}

server/kafka_message_persist.go

@@ -254,6 +254,9 @@ func prepareKafkaMessagePersistItem(ctx context.Context, message kafka.Message,
 		return nil, false, err
 	}
 	if !claimed {
+		if consume != nil && consume.Status != types.KafkaConsumeSucceeded && consume.Status != types.KafkaConsumeDead {
+			return nil, false, fmt.Errorf("%w: event=%s status=%s", errKafkaConsumeInProgress, envelope.EventID, consume.Status)
+		}
 		return nil, true, nil
 	}
 	item := &kafkaMessagePersistItem{envelope: envelope, tenantID: tenantID}
@@ -311,6 +314,8 @@ func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePers
 		// preparing the item.
 		immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "retry_exhausted").Inc()
 		immetrics.MessagePersistDuration.WithLabelValues(kind, "retry_exhausted").Observe(time.Since(started).Seconds())
+		logs.Err.Printf("kafka message persist retry exhausted: tenant=%d event=%s topic=%s seq=%d kind=%s err=%v",
+			item.tenantID, item.envelope.EventID, item.payload.Message.Topic, item.payload.Message.SeqId, kind, err)
 	}
 	return err
 }
@@ -341,8 +346,16 @@ func finishKafkaMessagePersist(ctx context.Context, item *kafkaMessagePersistIte
 			return fmt.Errorf("file lifecycle enqueue failed: %w", err)
 		}
 	}
-	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(item.tenantID, item.envelope.EventID,
+	if err := store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(item.tenantID, item.envelope.EventID,
-		kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
+		kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)); err != nil {
+		if errors.Is(err, types.ErrNotFound) {
+			logs.Warn.Printf("kafka message persist consume state missing after durable write: tenant=%d event=%s topic=%s seq=%d",
+				item.tenantID, item.envelope.EventID, item.payload.Message.Topic, item.payload.Message.SeqId)
+			return nil
+		}
+		return err
+	}
+	return nil
 }
 
 func kafkaMessagePersistFailureReason(err error) string {

server/kafka_runtime_batch_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/segmentio/kafka-go"
+	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
 
@@ -253,3 +254,30 @@ func TestRunKafkaMessageSingleRetriesStopsOnSuccess(t *testing.T) {
 		t.Fatalf("calls operation=%d failed=%d, want 3/2", operationCalls, markFailedCalls)
 	}
 }
+
+func TestFinishKafkaMessagePersistTreatsMissingConsumeStateAsSucceeded(t *testing.T) {
+	oldInfrastructure := store.HTTPInfrastructure
+	oldKafkaBus := kafkaBus
+	defer func() {
+		store.HTTPInfrastructure = oldInfrastructure
+		kafkaBus = oldKafkaBus
+	}()
+
+	store.HTTPInfrastructure = &testHTTPInfrastructureStore{markSucceededErr: types.ErrNotFound}
+	kafkaBus = &kafkaRuntime{config: kafkaConfig{ConsumerGroup: "test-consumer"}}
+
+	err := finishKafkaMessagePersist(context.Background(), &kafkaMessagePersistItem{
+		envelope: kafkaEnvelope{EventID: "msg.persist.8.p2pA.1"},
+		tenantID: types.TenantID(8),
+		payload: kafkaMessagePersistPayload{
+			MessageKind: "p2p",
+			Message: types.Message{
+				Topic: "p2pA",
+				SeqId: 1,
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("finishKafkaMessagePersist() error = %v, want nil", err)
+	}
+}