package automation import ( "context" "database/sql" "errors" "fmt" "net/http" "strings" "testing" "time" "github.com/nexuscrm/shared/pkg/models" ) func connectorPushTestIdentity() ConnectorPushDurableIdentity { return ConnectorPushDurableIdentity{ConnectorOwnerKey: "owner-key", ActionSlotID: "action-slot", ActionType: "PushToConnector", SourceObject: "Contact", SourceRecordID: "003-local", DestinationSlotID: "dest-slot"} } func connectorPushTestSnapshot() ConnectorPushDurableSnapshot { snapshot := ConnectorPushDurableSnapshot{RequesterOwnerPack: "requester-pack", ProfileOwnerPack: "profile-pack", ConnectorNamespace: "connector_ns", FlowDeveloperName: "Flow_API", FlowActionName: "Push_Action", TargetObject: "remoteContact", ProfileName: "default", CreatedPackInstallID: "install-1", CredentialRef: "nc-main", EndpointClass: "crm.objects.write", RetryPolicy: ConnectorPushRetryPolicy{MaxAttempts: 1}, CreatePathTemplate: "/objects", UpdatePathTemplate: "/objects/{remoteId}", ResponseIDPath: "id", MappingFingerprint: "mapping-fp", BodyFingerprint: "body-fp", OriginatingUserID: "005-user", OriginatingCredentialID: "session-1", OriginatingUserSession: models.UserSession{ID: "005-user", ProfileID: "profile-1"}, AllowedSourceFields: []string{"Name"}, RequestIdempotencyKey: "idem-2", RequestMethod: "POST", RequestRelativePath: "/objects", RequestHeaders: map[string][]string{"Content-Type": {"application/json"}}, RequestBody: []byte(`{"Name":"Ada"}`), RequestBodyHash: connectorPushTestRequestBodyHash} snapshot.ProfileContractHash = ConnectorPushProfileContractHash(snapshot) return snapshot } const connectorPushTestRequestBodyHash = "51b8ac54c01a8c66753a6404c4f0bc1633fee3c783f6152be76d9be196a50d44" func mustConnectorPushTestLinkKey(identity ConnectorPushDurableIdentity) string { linkKey, err := identity.LinkKey() if err != nil { panic(err) } return linkKey } // --- Inbound remote-id authorship evidence tests --- // testConnectorPushTxExtractor is a minimal txExtractor for testing in-tx insert. type testConnectorPushTxExtractor struct{ tx *sql.Tx } func (e *testConnectorPushTxExtractor) ExtractTx(_ context.Context) interface{} { return e.tx } func connectorInboundTestFinalization() ConnectorInboundRemoteIdentityFinalization { identity := connectorPushTestIdentity() snapshot := connectorPushTestSnapshot() snapshot.RemoteIDField = "HubSpot_Id__c" snapshot.ProfileContractHash = ConnectorPushProfileContractHash(snapshot) return ConnectorInboundRemoteIdentityFinalization{ Evidence: ConnectorInboundRemoteIDEvidence{ ID: "ev-inbound-finalize", ConnectorOwnerKey: identity.ConnectorOwnerKey, ConnectorNamespace: "hubspot-routing-namespace", SourceObject: identity.SourceObject, SourceRecordId: identity.SourceRecordID, RemoteIDField: snapshot.RemoteIDField, RemoteId: "remote-1", }, Adoption: ConnectorPushLinkAdoption{ LinkKey: mustConnectorPushTestLinkKey(identity), RemoteID: "remote-1", Identity: identity, Snapshot: snapshot, }, } } func TestSQLConnectorInboundFinalizeSupersedesExactHistoricalTerminalNoIOCreate(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } if _, err := db.ExecContext(ctx, connectorRemoteLinkTestDDL); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } finalization := connectorInboundTestFinalization() reservation := legacyConnectorAttemptFixture{ AttemptID: "attempt-historical-no-io", LinkKey: finalization.Adoption.LinkKey, AttemptOrdinal: 1, Identity: finalization.Adoption.Identity, Snapshot: connectorPushTestSnapshot(), } store := NewSQLConnectorPushDurableStore(db) if _, err := seedLegacyConnectorAttempt(ctx, db, reservation); err != nil { t.Fatalf("reserve historical create attempt: %v", err) } if _, err := db.ExecContext(ctx, ` UPDATE _System_ConnectorPushAttempt SET Status = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL, LeaseExpiresAt = NULL, RemoteIOStartedAt = NULL, ResponseStatusCode = NULL, RemoteId = NULL WHERE Id = ?`, ConnectorPushAttemptStatusTerminalFailed, reservation.AttemptID, ); err != nil { t.Fatalf("seed historical no-I/O terminal create: %v", err) } if err := store.InsertAuthoredEvidence(ctx, finalization.Evidence); err != nil { t.Fatalf("seed active authored evidence: %v", err) } var seededStatus, seededOccupancy, seededRemoteID string var seededResponseStatus, seededRemoteIOStarted int if err := db.QueryRowContext(ctx, ` SELECT Status, COALESCE(AttemptOccupancyKey, ''), COALESCE(ResponseStatusCode, -1), CASE WHEN RemoteIOStartedAt IS NULL THEN 0 ELSE 1 END, COALESCE(RemoteId, '') FROM _System_ConnectorPushAttempt WHERE Id = ?`, reservation.AttemptID).Scan( &seededStatus, &seededOccupancy, &seededResponseStatus, &seededRemoteIOStarted, &seededRemoteID, ); err != nil { t.Fatalf("read seeded no-I/O terminal create: %v", err) } if seededStatus != ConnectorPushAttemptStatusTerminalFailed || seededOccupancy != reservation.LinkKey || seededResponseStatus != -1 || seededRemoteIOStarted != 0 || seededRemoteID != "" { t.Fatalf("seeded no-I/O terminal create = status %q occupancy %q response %d remoteIOStarted %d remoteID %q", seededStatus, seededOccupancy, seededResponseStatus, seededRemoteIOStarted, seededRemoteID) } tx, err := db.Begin() if err != nil { t.Fatalf("begin finalization tx: %v", err) } ambientStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx}) if err := ambientStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { _ = tx.Rollback() t.Fatalf("finalize exact historical no-I/O terminal create: %v", err) } if err := tx.Commit(); err != nil { t.Fatalf("commit no-I/O terminal create finalization: %v", err) } var attemptStatus, occupancy, linkStatus, evidenceStatus string if err := db.QueryRowContext(ctx, ` SELECT Status, COALESCE(AttemptOccupancyKey, '') FROM _System_ConnectorPushAttempt WHERE Id = ?`, reservation.AttemptID).Scan(&attemptStatus, &occupancy); err != nil { t.Fatalf("read no-I/O attempt: %v", err) } if err := db.QueryRowContext(ctx, ` SELECT COALESCE(MAX(Status), '') FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, finalization.Adoption.LinkKey).Scan(&linkStatus); err != nil { t.Fatalf("read finalized remote link: %v", err) } if err := db.QueryRowContext(ctx, ` SELECT EvidenceStatus FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?`, finalization.Evidence.ID).Scan(&evidenceStatus); err != nil { t.Fatalf("read consumed evidence: %v", err) } if attemptStatus != ConnectorPushAttemptStatusSuperseded || occupancy != "" || linkStatus != ConnectorPushLinkStatusFinalized || evidenceStatus != ConnectorInboundEvidenceStatusConsumed { t.Fatalf("no-I/O finalization state = attempt %q/%q link %q evidence %q", attemptStatus, occupancy, linkStatus, evidenceStatus) } } func TestSQLConnectorInboundFinalizeHistoricalTerminalPostConflictIsAtomicAndIdempotent(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } if _, err := db.ExecContext(ctx, connectorRemoteLinkTestDDL); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } finalization := connectorInboundTestFinalization() reservation := legacyConnectorAttemptFixture{ AttemptID: "attempt-historical-409-atomic", LinkKey: finalization.Adoption.LinkKey, AttemptOrdinal: 1, Identity: finalization.Adoption.Identity, Snapshot: connectorPushTestSnapshot(), } store := NewSQLConnectorPushDurableStore(db) if _, err := seedLegacyConnectorAttempt(ctx, db, reservation); err != nil { t.Fatalf("reserve historical create attempt: %v", err) } if _, err := db.ExecContext(ctx, ` UPDATE _System_ConnectorPushAttempt SET Status = ?, ResponseStatusCode = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL, LeaseExpiresAt = NULL WHERE Id = ?`, ConnectorPushAttemptStatusTerminalFailed, http.StatusConflict, reservation.AttemptID, ); err != nil { t.Fatalf("seed historical terminal POST/409: %v", err) } if err := store.InsertAuthoredEvidence(ctx, finalization.Evidence); err != nil { t.Fatalf("seed active authored evidence: %v", err) } readState := func(t *testing.T) (attemptStatus, occupancy, supersededAt, linkStatus, evidenceStatus, consumedBy, activeKey string) { t.Helper() if err := db.QueryRowContext(ctx, ` SELECT Status, COALESCE(AttemptOccupancyKey, ''), COALESCE(CAST(SupersededAt AS CHAR), '') FROM _System_ConnectorPushAttempt WHERE Id = ?`, reservation.AttemptID).Scan(&attemptStatus, &occupancy, &supersededAt); err != nil { t.Fatalf("read attempt state: %v", err) } if err := db.QueryRowContext(ctx, ` SELECT COALESCE(MAX(Status), '') FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, finalization.Adoption.LinkKey).Scan(&linkStatus); err != nil { t.Fatalf("read link state: %v", err) } if err := db.QueryRowContext(ctx, ` SELECT EvidenceStatus, COALESCE(ConsumedByLinkKey, ''), COALESCE(ActiveKey, '') FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?`, finalization.Evidence.ID).Scan(&evidenceStatus, &consumedBy, &activeKey); err != nil { t.Fatalf("read evidence state: %v", err) } return } // The finalizer mutates all three rows, but a caller rollback must restore // the historical occupancy and authored evidence and leave no link behind. txRollback, err := db.Begin() if err != nil { t.Fatalf("begin rollback tx: %v", err) } rollbackStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txRollback}) if err := rollbackStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { _ = txRollback.Rollback() t.Fatalf("finalize before rollback: %v", err) } if err := txRollback.Rollback(); err != nil { t.Fatalf("rollback historical recovery: %v", err) } status, occupancy, supersededAt, linkStatus, evidenceStatus, consumedBy, activeKey := readState(t) if status != ConnectorPushAttemptStatusTerminalFailed || occupancy != finalization.Adoption.LinkKey || supersededAt != "" || linkStatus != "" || evidenceStatus != ConnectorInboundEvidenceStatusAuthored || consumedBy != "" || activeKey == "" { t.Fatalf("rollback state = attempt %q/%q/%q link %q evidence %q/%q/%q", status, occupancy, supersededAt, linkStatus, evidenceStatus, consumedBy, activeKey) } // The same operation committed publishes one finalized link and consumes the // exact evidence while clearing only the eligible historical occupancy. txCommit, err := db.Begin() if err != nil { t.Fatalf("begin commit tx: %v", err) } commitStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txCommit}) if err := commitStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { _ = txCommit.Rollback() t.Fatalf("finalize before commit: %v", err) } if err := txCommit.Commit(); err != nil { t.Fatalf("commit historical recovery: %v", err) } status, occupancy, supersededAt, linkStatus, evidenceStatus, consumedBy, activeKey = readState(t) if status != ConnectorPushAttemptStatusSuperseded || occupancy != "" || supersededAt == "" || linkStatus != ConnectorPushLinkStatusFinalized || evidenceStatus != ConnectorInboundEvidenceStatusConsumed || consumedBy != finalization.Adoption.LinkKey || activeKey != "" { t.Fatalf("commit state = attempt %q/%q/%q link %q evidence %q/%q/%q", status, occupancy, supersededAt, linkStatus, evidenceStatus, consumedBy, activeKey) } // A retry names the same consumed evidence and exact finalized authority; it // must succeed without another link or supersession. txRetry, err := db.Begin() if err != nil { t.Fatalf("begin retry tx: %v", err) } retryStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txRetry}) if err := retryStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { _ = txRetry.Rollback() t.Fatalf("idempotent historical recovery retry: %v", err) } if err := txRetry.Commit(); err != nil { t.Fatalf("commit retry: %v", err) } var linkCount int if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, finalization.Adoption.LinkKey).Scan(&linkCount); err != nil { t.Fatalf("count links after retry: %v", err) } if linkCount != 1 { t.Fatalf("links after retry = %d, want 1", linkCount) } } func TestSQLConnectorInboundHistoricalTerminalPostConflictRecoveryRejectsIneligibleShapes(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorRemoteLinkTestDDL); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } cases := []struct { name string column string value any }{ {name: "pending status", column: "Status", value: ConnectorPushAttemptStatusPending}, {name: "ambiguous remote effect", column: "Status", value: ConnectorPushAttemptStatusAmbiguousRemoteEffect}, {name: "wrong method", column: "RequestMethod", value: http.MethodPatch}, {name: "non conflict response", column: "ResponseStatusCode", value: http.StatusInternalServerError}, {name: "remote id already materialized", column: "RemoteId", value: "remote-from-attempt"}, {name: "conflict resolution fingerprint", column: "CreateConflictResolutionFingerprint", value: "fingerprint"}, {name: "create conflict patch intent", column: "CreateConflictPatchIntentJSON", value: `{"requestBodyHash":"hash"}`}, {name: "patch dispatch intent", column: "PatchDispatchIntentJSON", value: `{"state":"ready"}`}, {name: "claimed worker", column: "ClaimedBy", value: "worker"}, {name: "claim token", column: "ClaimToken", value: "claim"}, {name: "claimed timestamp", column: "ClaimedAt", value: time.Now().UTC()}, {name: "active lease", column: "LeaseExpiresAt", value: time.Now().UTC().Add(time.Hour)}, {name: "successor attempt", column: "SupersededByAttemptId", value: "successor"}, {name: "prior supersession timestamp", column: "SupersededAt", value: time.Now().UTC()}, } for index, tc := range cases { t.Run(tc.name, func(t *testing.T) { identity := connectorPushTestIdentity() identity.SourceRecordID = fmt.Sprintf("003-historical-guard-%02d", index) linkKey := mustConnectorPushTestLinkKey(identity) snapshot := connectorPushTestSnapshot() snapshot.RequestIdempotencyKey = fmt.Sprintf("idem-historical-guard-%02d", index) attemptID := fmt.Sprintf("attempt-historical-guard-%02d", index) if _, err := seedLegacyConnectorAttempt(ctx, db, legacyConnectorAttemptFixture{ AttemptID: attemptID, LinkKey: linkKey, AttemptOrdinal: 1, Identity: identity, Snapshot: snapshot, }); err != nil { t.Fatalf("reserve attempt: %v", err) } if _, err := db.ExecContext(ctx, ` UPDATE _System_ConnectorPushAttempt SET Status = ?, ResponseStatusCode = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL, LeaseExpiresAt = NULL WHERE Id = ?`, ConnectorPushAttemptStatusTerminalFailed, http.StatusConflict, attemptID, ); err != nil { t.Fatalf("seed baseline terminal POST/409: %v", err) } if _, err := db.ExecContext(ctx, fmt.Sprintf("UPDATE _System_ConnectorPushAttempt SET %s = ? WHERE Id = ?", tc.column), tc.value, attemptID); err != nil { t.Fatalf("mutate ineligible shape: %v", err) } tx, err := db.Begin() if err != nil { t.Fatalf("begin recovery tx: %v", err) } recovered, err := supersedeExactHistoricalTerminalAttemptTx(ctx, tx, attemptID, linkKey) if err != nil { _ = tx.Rollback() t.Fatalf("evaluate ineligible recovery: %v", err) } if recovered { _ = tx.Rollback() t.Fatal("ineligible historical attempt was recovered") } if err := tx.Commit(); err != nil { t.Fatalf("commit no-op recovery tx: %v", err) } var status, occupancy string if err := db.QueryRowContext(ctx, ` SELECT Status, COALESCE(AttemptOccupancyKey, '') FROM _System_ConnectorPushAttempt WHERE Id = ?`, attemptID).Scan(&status, &occupancy); err != nil { t.Fatalf("read ineligible attempt: %v", err) } if status == ConnectorPushAttemptStatusSuperseded || occupancy != linkKey { t.Fatalf("ineligible attempt mutated to status/occupancy %q/%q", status, occupancy) } }) } } // connectorInboundEvidenceTestDDL mirrors the _System_ConnectorInboundRemoteIDEvidence // columns/keys from bootstrap/system_tables.json closely enough for the in-tx // lifecycle integration test below (PK on Id, UNIQUE on ActiveKey). const connectorInboundEvidenceTestDDL = ` CREATE TABLE _System_ConnectorInboundRemoteIDEvidence ( Id VARCHAR(255) NOT NULL, ConnectorOwnerKey VARCHAR(255) NOT NULL, ConnectorNamespace VARCHAR(255) NOT NULL, SourceObject VARCHAR(255) NOT NULL, SourceRecordId VARCHAR(255) NOT NULL, RemoteIDField VARCHAR(255) NOT NULL, RemoteId VARCHAR(512) NOT NULL, EvidenceStatus VARCHAR(32) NOT NULL, ConsumedByLinkKey VARCHAR(255) NULL, ActiveKey VARCHAR(255) NULL, CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (Id), UNIQUE KEY uniq_connector_inbound_evidence_active (ActiveKey) )` // TestSQLConnectorInboundEvidenceTransitionsAreAtomicWithAmbientTx is the real // DB-backed proof of the #2046 in-transaction lifecycle acceptance contract // (IANTHEREAL gate): when the store is wired with a tx-extractor, RetireEvidence // AND MarkEvidenceConsumed run inside the ambient *sql.Tx, so a ROLLBACK leaves // the rows still authored/active/unconsumed (the guarantee that prevents durable // authorship corruption when an inbound flow later rolls back), and a COMMIT // persists them. Skipped when TiDB is unreachable. func TestSQLConnectorInboundEvidenceTransitionsAreAtomicWithAmbientTx(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } const owner, ns, object, field = "owner-pack", "owner_ns", "Contact", "HubSpot_Id__c" const recRetire = "003-atomic-retire" plain := NewSQLConnectorPushDurableStore(db) seed := func(id, record, remote string) { if err := plain.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: id, ConnectorOwnerKey: owner, ConnectorNamespace: ns, SourceObject: object, SourceRecordId: record, RemoteIDField: field, RemoteId: remote, }); err != nil { t.Fatalf("seed authored evidence %s: %v", id, err) } } seed("ev-atomic-retire", recRetire, "remote-retire") mustActive := func(record, wantConsumedBy string) *ConnectorInboundRemoteIDEvidence { got, err := plain.FindActiveInboundRemoteIDEvidence(ctx, owner, object, record, field) if err != nil { t.Fatalf("find active (%s): %v", record, err) } if got == nil { t.Fatalf("expected (%s) to still be active authored, got nil", record) } if got.EvidenceStatus != ConnectorInboundEvidenceStatusAuthored { t.Fatalf("(%s) status = %q, want authored", record, got.EvidenceStatus) } if got.ConsumedByLinkKey != wantConsumedBy { t.Fatalf("(%s) ConsumedByLinkKey = %q, want %q", record, got.ConsumedByLinkKey, wantConsumedBy) } return got } // --- ROLLBACK case: Retire runs inside the tx; rollback must undo it. --- txRB, err := db.Begin() if err != nil { t.Fatalf("begin rollback tx: %v", err) } rbStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txRB}) if err := rbStore.RetireEvidence(ctx, "ev-atomic-retire"); err != nil { t.Fatalf("RetireEvidence in tx: %v", err) } if err := txRB.Rollback(); err != nil { t.Fatalf("rollback: %v", err) } // Real read-back: the row must be untouched (rollback undid the in-tx writes). mustActive(recRetire, "") // --- COMMIT case: same transitions, committed, must PERSIST. --- txC, err := db.Begin() if err != nil { t.Fatalf("begin commit tx: %v", err) } cStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txC}) if err := cStore.RetireEvidence(ctx, "ev-atomic-retire"); err != nil { t.Fatalf("RetireEvidence in commit tx: %v", err) } if err := txC.Commit(); err != nil { t.Fatalf("commit: %v", err) } // After commit, the retired row is no longer active (ActiveKey cleared). if got, _ := plain.FindActiveInboundRemoteIDEvidence(ctx, owner, object, recRetire, field); got != nil { t.Fatalf("commit must persist retire; row still active: %#v", got) } // Confirm the persisted terminal states via raw read. var retireStatus string if err := db.QueryRowContext(ctx, "SELECT EvidenceStatus FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?", "ev-atomic-retire").Scan(&retireStatus); err != nil { t.Fatalf("read retired status: %v", err) } if retireStatus != ConnectorInboundEvidenceStatusRetired { t.Fatalf("committed retire status = %q, want retired", retireStatus) } } // TestSQLConnectorInboundFindInTxObservesUncommittedAmbientWrite is the real // DB-backed proof of the register action's in-tx idempotency-read fix (Gate 0 // P1): FindActiveInboundRemoteIDEvidenceInTx must read through the ambient *sql.Tx // so it observes evidence the SAME tx wrote but has not committed yet // (read-after-write), while the *sql.DB-backed FindActiveInboundRemoteIDEvidence // must NOT see that uncommitted row (it is reserved for the cross-tx conflict // reread). Skipped when TiDB is unreachable. func TestSQLConnectorInboundFindInTxObservesUncommittedAmbientWrite(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } const owner, ns, object, field, record = "owner-pack", "owner_ns", "Contact", "HubSpot_Id__c", "003-intx-read" tx, err := db.Begin() if err != nil { t.Fatalf("begin tx: %v", err) } txStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx}) plain := NewSQLConnectorPushDurableStore(db) // Write evidence inside the ambient tx (uncommitted). if err := txStore.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: "ev-intx-read", ConnectorOwnerKey: owner, ConnectorNamespace: ns, SourceObject: object, SourceRecordId: record, RemoteIDField: field, RemoteId: "remote-intx", }); err != nil { t.Fatalf("InsertAuthoredEvidence in tx: %v", err) } // In-tx read MUST observe the tx's own uncommitted write (the fix). got, err := txStore.FindActiveInboundRemoteIDEvidenceInTx(ctx, owner, object, record, field) if err != nil { t.Fatalf("FindActiveInboundRemoteIDEvidenceInTx: %v", err) } if got == nil || got.RemoteId != "remote-intx" { t.Fatalf("in-tx read must see the uncommitted ambient write; got %#v", got) } // A *sql.DB read (no tx) MUST NOT see the uncommitted row yet. if dbGot, err := plain.FindActiveInboundRemoteIDEvidence(ctx, owner, object, record, field); err != nil { t.Fatalf("plain find before commit: %v", err) } else if dbGot != nil { t.Fatalf("*sql.DB read must NOT see the uncommitted in-tx write; got %#v", dbGot) } if err := tx.Commit(); err != nil { t.Fatalf("commit: %v", err) } // After commit, the *sql.DB read sees it. if dbGot, err := plain.FindActiveInboundRemoteIDEvidence(ctx, owner, object, record, field); err != nil { t.Fatalf("plain find after commit: %v", err) } else if dbGot == nil || dbGot.RemoteId != "remote-intx" { t.Fatalf("after commit the *sql.DB read must see the row; got %#v", dbGot) } } func TestSQLConnectorTransferInboundRemoteIdentityMovesEvidenceAndRetiresSourceLink(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } if _, err := db.ExecContext(ctx, strings.Replace(connectorRemoteLinkTestDDL, "RemoteLocatorActiveKey VARCHAR(255) NOT NULL", "RemoteLocatorActiveKey VARCHAR(255) NULL", 1)); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } const ( owner = "source-pack" ns = "connector_ns" field = "External_Id__c" remote = "remote-229009332242" srcID = "00Q-transfer-source" tgtID = "003-transfer-target" ) store := NewSQLConnectorPushDurableStore(db) if err := store.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: "ev-transfer-source", ConnectorOwnerKey: owner, ConnectorNamespace: ns, SourceObject: "Lead", SourceRecordId: srcID, RemoteIDField: field, RemoteId: remote, }); err != nil { t.Fatalf("seed source evidence: %v", err) } sourceIdentity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "action-slot", ActionType: "PushToConnector", SourceObject: "Lead", SourceRecordID: srcID, DestinationSlotID: "destination-slot", } sourceLinkKey, err := sourceIdentity.LinkKey() if err != nil { t.Fatalf("source link key: %v", err) } insertSyncLink(t, db, "link-transfer-source", sourceLinkKey, ConnectorPushLinkStatusFinalized, remote, owner, sourceIdentity.ActionSlotID, sourceIdentity.ActionType, sourceIdentity.SourceObject, sourceIdentity.SourceRecordID, sourceIdentity.DestinationSlotID, "target-object") if _, err := db.ExecContext(ctx, ` UPDATE _System_ConnectorRemoteLink SET RemoteLocatorHash = ?, RemoteLocatorActiveKey = ?, RemoteIdField = ?, ConnectorNamespace = ? WHERE LinkKey = ?`, ConnectorPushRemoteLocatorHash(sourceIdentity, remote), ConnectorPushRemoteLocatorHash(sourceIdentity, remote), field, ns, sourceLinkKey, ); err != nil { t.Fatalf("set source link remote locator: %v", err) } tx, err := db.Begin() if err != nil { t.Fatalf("begin tx: %v", err) } txStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx}) transfer := ConnectorRemoteIdentityTransfer{ SourceObject: "Lead", SourceRecordId: srcID, TargetObject: "Contact", TargetRecordId: tgtID, RemoteIDField: field, RemoteId: remote, } transfer.admissions = []connectorRemoteIdentityTransferAdmission{{ ConnectorNamespace: ns, RemoteID: remote, SourceIdentity: sourceIdentity, TargetIdentity: ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "target-action-slot", ActionType: sourceIdentity.ActionType, SourceObject: "Contact", SourceRecordID: tgtID, DestinationSlotID: sourceIdentity.DestinationSlotID, }, }} if gotRemoteID, err := txStore.TransferInboundRemoteIdentity(ctx, connectorTransferFixtureSnapshots(transfer)); err != nil { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity: %v", err) } else if gotRemoteID != remote { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity remoteID = %q, want %q", gotRemoteID, remote) } if err := tx.Commit(); err != nil { t.Fatalf("commit: %v", err) } if got, err := store.FindActiveInboundRemoteIDEvidence(ctx, owner, "Contact", tgtID, field); err != nil { t.Fatalf("find target evidence: %v", err) } else if got != nil { t.Fatalf("conversion fabricated target inbound evidence: %#v", got) } if got, err := store.FindActiveInboundRemoteIDEvidence(ctx, owner, "Lead", srcID, field); err != nil { t.Fatalf("find source evidence: %v", err) } else if got != nil { t.Fatalf("source evidence still active after transfer: %#v", got) } var linkStatus, activeLocator string if err := db.QueryRowContext(ctx, `SELECT Status, COALESCE(RemoteLocatorActiveKey, '') FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, sourceLinkKey).Scan(&linkStatus, &activeLocator); err != nil { t.Fatalf("read source link after transfer: %v", err) } if linkStatus != ConnectorPushLinkStatusRetired || activeLocator != "" { t.Fatalf("source link status/active locator = %q/%q, want retired/blank", linkStatus, activeLocator) } // Transfer itself publishes the sole active finalized target. A later // inbound flow sees existing authority without requiring a fabricated write. targetIdentity := transfer.admissions[0].TargetIdentity targetTx, err := db.Begin() if err != nil { t.Fatalf("begin target finalization: %v", err) } targetStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: targetTx}) if err := targetStore.LockConnectorRemoteIdentityLaneInAmbientTx(ctx, targetIdentity, remote); err != nil { _ = targetTx.Rollback() t.Fatalf("lock target locator lane: %v", err) } if already, err := targetStore.CheckConnectorRemoteIdentityRegistrationInAmbientTx(ctx, targetIdentity, remote); err != nil { _ = targetTx.Rollback() t.Fatalf("active-only admission after transfer: %v", err) } else if !already { _ = targetTx.Rollback() t.Fatal("transferred target is not already finalized") } if err := targetTx.Commit(); err != nil { t.Fatalf("commit target finalization: %v", err) } var activeCount, retiredCount int if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE RemoteLocatorHash = ? AND Status = ? AND RemoteLocatorActiveKey = ?`, ConnectorPushRemoteLocatorHash(targetIdentity, remote), ConnectorPushLinkStatusFinalized, ConnectorPushRemoteLocatorHash(targetIdentity, remote)).Scan(&activeCount); err != nil { t.Fatalf("count active target authority: %v", err) } if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE RemoteLocatorHash = ? AND Status = ? AND RemoteLocatorActiveKey IS NULL`, ConnectorPushRemoteLocatorHash(sourceIdentity, remote), ConnectorPushLinkStatusRetired).Scan(&retiredCount); err != nil { t.Fatalf("count retired source lineage: %v", err) } if activeCount != 1 || retiredCount != 1 { t.Fatalf("locator lineage active/retired = %d/%d, want 1/1", activeCount, retiredCount) } } func TestSQLConnectorTransferInboundRemoteIdentityUsesSourceLinkAuthority(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } if _, err := db.ExecContext(ctx, strings.Replace(connectorRemoteLinkTestDDL, "RemoteLocatorActiveKey VARCHAR(255) NOT NULL", "RemoteLocatorActiveKey VARCHAR(255) NULL", 1)); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } const ( owner = "source-pack" ns = "connector_ns" field = "External_Id__c" remote = "remote-229009332242" srcID = "00Q-transfer-link-only" tgtID = "003-transfer-link-only" ) store := NewSQLConnectorPushDurableStore(db) sourceIdentity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "action-slot", ActionType: "PushToConnector", SourceObject: "Lead", SourceRecordID: srcID, DestinationSlotID: "destination-slot", } sourceLinkKey, err := sourceIdentity.LinkKey() if err != nil { t.Fatalf("source link key: %v", err) } insertSyncLink(t, db, "link-transfer-link-only", sourceLinkKey, ConnectorPushLinkStatusFinalized, remote, owner, sourceIdentity.ActionSlotID, sourceIdentity.ActionType, sourceIdentity.SourceObject, sourceIdentity.SourceRecordID, sourceIdentity.DestinationSlotID, "target-object") if _, err := db.ExecContext(ctx, ` UPDATE _System_ConnectorRemoteLink SET RemoteLocatorHash = ?, RemoteLocatorActiveKey = ?, RemoteIdField = ?, ConnectorNamespace = ? WHERE LinkKey = ?`, ConnectorPushRemoteLocatorHash(sourceIdentity, remote), ConnectorPushRemoteLocatorHash(sourceIdentity, remote), field, ns, sourceLinkKey, ); err != nil { t.Fatalf("set source link remote locator: %v", err) } tx, err := db.Begin() if err != nil { t.Fatalf("begin tx: %v", err) } txStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx}) transfer := ConnectorRemoteIdentityTransfer{ SourceObject: "Lead", SourceRecordId: srcID, TargetObject: "Contact", TargetRecordId: tgtID, RemoteIDField: field, } transfer.admissions = []connectorRemoteIdentityTransferAdmission{{ ConnectorNamespace: ns, RemoteID: remote, SourceIdentity: sourceIdentity, TargetIdentity: ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "target-action-slot", ActionType: sourceIdentity.ActionType, SourceObject: "Contact", SourceRecordID: tgtID, DestinationSlotID: sourceIdentity.DestinationSlotID, }, }} if gotRemoteID, err := txStore.TransferInboundRemoteIdentity(ctx, connectorTransferFixtureSnapshots(transfer)); err != nil { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity: %v", err) } else if gotRemoteID != remote { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity remoteID = %q, want %q", gotRemoteID, remote) } if err := tx.Commit(); err != nil { t.Fatalf("commit: %v", err) } if got, err := store.FindActiveInboundRemoteIDEvidence(ctx, owner, "Contact", tgtID, field); err != nil { t.Fatalf("find target evidence: %v", err) } else if got != nil { t.Fatalf("conversion fabricated target inbound evidence: %#v", got) } if got, err := store.FindFinalizedLink(ctx, transfer.admissions[0].TargetIdentity); err != nil || got == nil || got.RemoteID != remote { t.Fatalf("target finalized identity = %#v, %v", got, err) } var linkStatus, activeLocator string if err := db.QueryRowContext(ctx, `SELECT Status, COALESCE(RemoteLocatorActiveKey, '') FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, sourceLinkKey).Scan(&linkStatus, &activeLocator); err != nil { t.Fatalf("read source link after transfer: %v", err) } if linkStatus != ConnectorPushLinkStatusRetired || activeLocator != "" { t.Fatalf("source link status/active locator = %q/%q, want retired/blank", linkStatus, activeLocator) } } func TestSQLConnectorTransferInboundRemoteIdentityRejectsTargetAuthorityWithoutSourceAuthority(t *testing.T) { for _, tc := range []struct { name string targetRemoteID string wantConflict bool sourceAuthority bool seed func(t *testing.T, db *sql.DB, store *SQLConnectorPushDurableStore, targetID, field string) }{ { name: "authored evidence", targetRemoteID: "copied-mirror", wantConflict: true, seed: func(t *testing.T, _ *sql.DB, store *SQLConnectorPushDurableStore, targetID, field string) { t.Helper() if err := store.InsertAuthoredEvidence(context.Background(), ConnectorInboundRemoteIDEvidence{ ID: "ev-target-owned", ConnectorOwnerKey: "target-pack", ConnectorNamespace: "connector_ns", SourceObject: "Contact", SourceRecordId: targetID, RemoteIDField: field, RemoteId: "target-remote", }); err != nil { t.Fatalf("seed target evidence: %v", err) } }, }, { name: "foreign target evidence with source authority", targetRemoteID: "source-remote", wantConflict: true, sourceAuthority: true, seed: func(t *testing.T, _ *sql.DB, store *SQLConnectorPushDurableStore, targetID, field string) { t.Helper() if err := store.InsertAuthoredEvidence(context.Background(), ConnectorInboundRemoteIDEvidence{ ID: "ev-foreign-target-owned", ConnectorOwnerKey: "foreign-target-pack", ConnectorNamespace: "foreign_ns", SourceObject: "Contact", SourceRecordId: targetID, RemoteIDField: field, RemoteId: "foreign-remote", }); err != nil { t.Fatalf("seed foreign target evidence: %v", err) } }, }, { name: "finalized link", targetRemoteID: "copied-mirror", wantConflict: true, seed: func(t *testing.T, db *sql.DB, _ *SQLConnectorPushDurableStore, targetID, field string) { t.Helper() identity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: "target-pack", ActionSlotID: "target-slot", ActionType: "PushToConnector", SourceObject: "Contact", SourceRecordID: targetID, DestinationSlotID: "target-destination", } linkKey, err := identity.LinkKey() if err != nil { t.Fatalf("target LinkKey: %v", err) } insertSyncLink(t, db, "link-target-owned", linkKey, ConnectorPushLinkStatusFinalized, "target-remote", identity.ConnectorOwnerKey, identity.ActionSlotID, identity.ActionType, identity.SourceObject, identity.SourceRecordID, identity.DestinationSlotID, "contacts") if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET RemoteIdField = ? WHERE LinkKey = ?`, field, linkKey); err != nil { t.Fatalf("set target link remote-id field: %v", err) } }, }, { name: "finalized link with blank target mirror", wantConflict: true, seed: func(t *testing.T, db *sql.DB, _ *SQLConnectorPushDurableStore, targetID, field string) { t.Helper() identity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: "target-pack", ActionSlotID: "target-slot", ActionType: "PushToConnector", SourceObject: "Contact", SourceRecordID: targetID, DestinationSlotID: "target-destination", } linkKey, err := identity.LinkKey() if err != nil { t.Fatalf("target LinkKey: %v", err) } insertSyncLink(t, db, "link-target-owned-blank-mirror", linkKey, ConnectorPushLinkStatusFinalized, "target-remote", identity.ConnectorOwnerKey, identity.ActionSlotID, identity.ActionType, identity.SourceObject, identity.SourceRecordID, identity.DestinationSlotID, "contacts") if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET RemoteIdField = ? WHERE LinkKey = ?`, field, linkKey); err != nil { t.Fatalf("set target link remote-id field: %v", err) } }, }, { name: "owner deleted pending recovery link", targetRemoteID: "copied-mirror", wantConflict: true, seed: func(t *testing.T, db *sql.DB, _ *SQLConnectorPushDurableStore, targetID, field string) { t.Helper() identity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: "target-pack", ActionSlotID: "target-slot", ActionType: "PushToConnector", SourceObject: "Contact", SourceRecordID: targetID, DestinationSlotID: "target-destination", } linkKey, err := identity.LinkKey() if err != nil { t.Fatalf("target LinkKey: %v", err) } insertSyncLink(t, db, "link-target-pending-recovery", linkKey, ConnectorPushLinkStatusOwnerDeletedPendingRecovery, "target-remote", identity.ConnectorOwnerKey, identity.ActionSlotID, identity.ActionType, identity.SourceObject, identity.SourceRecordID, identity.DestinationSlotID, "contacts") if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET RemoteIdField = ? WHERE LinkKey = ?`, field, linkKey); err != nil { t.Fatalf("set target link remote-id field: %v", err) } }, }, } { t.Run(tc.name, func(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } if _, err := db.ExecContext(ctx, strings.Replace(connectorRemoteLinkTestDDL, "RemoteLocatorActiveKey VARCHAR(255) NOT NULL", "RemoteLocatorActiveKey VARCHAR(255) NULL", 1)); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } const ( sourceID = "00Q-transfer-no-authority" targetID = "003-transfer-target-authority" field = "External_Id__c" ) store := NewSQLConnectorPushDurableStore(db) tc.seed(t, db, store, targetID, field) tx, err := db.Begin() if err != nil { t.Fatalf("begin: %v", err) } txStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx}) transfer := ConnectorRemoteIdentityTransfer{ SourceObject: "Lead", SourceRecordId: sourceID, TargetObject: "Contact", TargetRecordId: targetID, RemoteIDField: field, TargetRemoteID: tc.targetRemoteID, } if tc.sourceAuthority { if err := store.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: "ev-source-authority", ConnectorOwnerKey: "source-pack", ConnectorNamespace: "source_ns", SourceObject: "Lead", SourceRecordId: sourceID, RemoteIDField: field, RemoteId: "source-remote", }); err != nil { _ = tx.Rollback() t.Fatalf("seed source evidence: %v", err) } sourceIdentity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: "source-pack", ActionSlotID: "source-slot", ActionType: "PushToConnector", SourceObject: "Lead", SourceRecordID: sourceID, DestinationSlotID: "source-destination", } transfer.RemoteId = "source-remote" transfer.admissions = []connectorRemoteIdentityTransferAdmission{{ ConnectorNamespace: "source_ns", RemoteID: "source-remote", SourceIdentity: sourceIdentity, TargetIdentity: ConnectorPushDurableIdentity{ ConnectorOwnerKey: "source-pack", ActionSlotID: "target-slot", ActionType: "PushToConnector", SourceObject: "Contact", SourceRecordID: targetID, DestinationSlotID: "source-destination", }, }} } _, err = txStore.TransferInboundRemoteIdentity(ctx, connectorTransferFixtureSnapshots(transfer)) if tc.wantConflict && !errors.Is(err, ErrConnectorPushDurableConflict) { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity error = %v, want durable conflict", err) } if !tc.wantConflict && err != nil { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity error = %v, want nil", err) } if err := tx.Rollback(); err != nil { t.Fatalf("rollback: %v", err) } }) } } func TestSQLConnectorTransferInboundRemoteIdentityCoalescesSameRemoteIDWithNamespaceEcho(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() ctx := context.Background() if _, err := db.ExecContext(ctx, connectorInboundEvidenceTestDDL); err != nil { t.Fatalf("create evidence table: %v", err) } if _, err := db.ExecContext(ctx, strings.Replace(connectorRemoteLinkTestDDL, "RemoteLocatorActiveKey VARCHAR(255) NOT NULL", "RemoteLocatorActiveKey VARCHAR(255) NULL", 1)); err != nil { t.Fatalf("create remote link table: %v", err) } createConnectorRuntimeOwnerCatalogTables(t, db) if _, err := db.ExecContext(ctx, connectorPushAttemptTestDDL); err != nil { t.Fatalf("create attempt table: %v", err) } const ( owner = "pingcap-hubspot-companion" linkNS = "pingcap-hubspot-companion" evidenceNS = "hubspot-crm-connector" field = "HubSpot_Id__c" remote = "231388692800" srcID = "00Q-transfer-echo-source" tgtID = "003-transfer-echo-target" ) store := NewSQLConnectorPushDurableStore(db) if err := store.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: "ev-transfer-echo-source", ConnectorOwnerKey: owner, ConnectorNamespace: evidenceNS, SourceObject: "Lead", SourceRecordId: srcID, RemoteIDField: field, RemoteId: remote, }); err != nil { t.Fatalf("seed source evidence: %v", err) } sourceIdentity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "hubspot-contact", ActionType: "PushToConnector", SourceObject: "Lead", SourceRecordID: srcID, DestinationSlotID: "hubspot-contact", } sourceLinkKey, err := sourceIdentity.LinkKey() if err != nil { t.Fatalf("source link key: %v", err) } insertSyncLink(t, db, "link-transfer-echo-source", sourceLinkKey, ConnectorPushLinkStatusFinalized, remote, owner, sourceIdentity.ActionSlotID, sourceIdentity.ActionType, sourceIdentity.SourceObject, sourceIdentity.SourceRecordID, sourceIdentity.DestinationSlotID, "contacts") if _, err := db.ExecContext(ctx, ` UPDATE _System_ConnectorRemoteLink SET RemoteLocatorHash = ?, RemoteLocatorActiveKey = ?, RemoteIdField = ?, ConnectorNamespace = ? WHERE LinkKey = ?`, ConnectorPushRemoteLocatorHash(sourceIdentity, remote), ConnectorPushRemoteLocatorHash(sourceIdentity, remote), field, linkNS, sourceLinkKey, ); err != nil { t.Fatalf("set source link remote locator: %v", err) } tx, err := db.Begin() if err != nil { t.Fatalf("begin tx: %v", err) } txStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: tx}) transfer := ConnectorRemoteIdentityTransfer{ SourceObject: "Lead", SourceRecordId: srcID, TargetObject: "Contact", TargetRecordId: tgtID, RemoteIDField: field, } transfer.admissions = []connectorRemoteIdentityTransferAdmission{{ ConnectorNamespace: evidenceNS, RemoteID: remote, SourceIdentity: sourceIdentity, TargetIdentity: ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "hubspot-contact-target", ActionType: sourceIdentity.ActionType, SourceObject: "Contact", SourceRecordID: tgtID, DestinationSlotID: sourceIdentity.DestinationSlotID, }, }} if gotRemoteID, err := txStore.TransferInboundRemoteIdentity(ctx, connectorTransferFixtureSnapshots(transfer)); err != nil { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity: %v", err) } else if gotRemoteID != remote { _ = tx.Rollback() t.Fatalf("TransferInboundRemoteIdentity remoteID = %q, want %q", gotRemoteID, remote) } if err := tx.Commit(); err != nil { t.Fatalf("commit: %v", err) } if got, err := store.FindActiveInboundRemoteIDEvidence(ctx, owner, "Contact", tgtID, field); err != nil { t.Fatalf("find target evidence: %v", err) } else if got != nil { t.Fatalf("conversion fabricated target inbound evidence: %#v", got) } if got, err := store.FindFinalizedLink(ctx, transfer.admissions[0].TargetIdentity); err != nil || got == nil || got.RemoteID != remote { t.Fatalf("target finalized identity = %#v, %v", got, err) } if got, err := store.FindActiveInboundRemoteIDEvidence(ctx, owner, "Lead", srcID, field); err != nil { t.Fatalf("find source evidence: %v", err) } else if got != nil { t.Fatalf("source evidence still active after transfer: %#v", got) } var linkStatus, activeLocator string if err := db.QueryRowContext(ctx, `SELECT Status, COALESCE(RemoteLocatorActiveKey, '') FROM _System_ConnectorRemoteLink WHERE LinkKey = ?`, sourceLinkKey).Scan(&linkStatus, &activeLocator); err != nil { t.Fatalf("read source link after transfer: %v", err) } if linkStatus != ConnectorPushLinkStatusRetired || activeLocator != "" { t.Fatalf("source link status/active locator = %q/%q, want retired/blank", linkStatus, activeLocator) } } // connectorRemoteLinkTestDDL is a minimal CREATE for the AdoptFinalizedLink // integration test — only the columns touched by AdoptFinalizedLink INSERT and // FindLink SELECT are listed. const connectorRemoteLinkTestDDL = ` CREATE TABLE _System_ConnectorRemoteLink ( Id VARCHAR(255) NOT NULL, LinkKey VARCHAR(255) NOT NULL, Status VARCHAR(64) NOT NULL, RemoteId VARCHAR(512) NOT NULL, RemoteLocatorHash VARCHAR(255) NOT NULL, RemoteLocatorActiveKey VARCHAR(255) NOT NULL, ConnectorOwnerKey VARCHAR(255) NOT NULL, ActionSlotId VARCHAR(255) NOT NULL, ActionType VARCHAR(64) NOT NULL, SourceObject VARCHAR(255) NOT NULL, SourceRecordId VARCHAR(255) NOT NULL, DestinationSlotId VARCHAR(255) NOT NULL, RequesterOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ProfileOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ConnectorNamespace VARCHAR(255) NOT NULL DEFAULT '', FlowDeveloperName VARCHAR(255) NOT NULL DEFAULT '', FlowActionName VARCHAR(255) NOT NULL DEFAULT '', TargetObject VARCHAR(255) NOT NULL DEFAULT '', ProfileName VARCHAR(255) NOT NULL DEFAULT '', CreatedPackInstallId VARCHAR(255) NOT NULL DEFAULT '', LastValidatedPackInstallId VARCHAR(255) NOT NULL DEFAULT '', CredentialRef VARCHAR(255) NOT NULL DEFAULT '', EndpointClass VARCHAR(255) NOT NULL DEFAULT '', RetryPolicy LONGTEXT NOT NULL, ProfileContractHash VARCHAR(255) NOT NULL DEFAULT '', CreatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', UpdatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', ResponseIdPath VARCHAR(255) NOT NULL DEFAULT '', RemoteIdField VARCHAR(255) NULL, RemoteIdProvenanceField VARCHAR(255) NULL, MappingFingerprint VARCHAR(255) NOT NULL DEFAULT '', BodyFingerprint VARCHAR(255) NOT NULL DEFAULT '', CompatibilityMirrorStatus VARCHAR(64) NOT NULL DEFAULT '', CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (Id), UNIQUE KEY uniq_connector_remote_link_key (LinkKey), UNIQUE KEY uniq_connector_remote_locator_active (RemoteLocatorActiveKey), KEY idx_connector_remote_link_source (SourceObject, SourceRecordId) ) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci` // connectorPushAttemptTestDDL is a minimal CREATE for AdoptFinalizedLink's // AttemptOccupancyKey FOR UPDATE gate. const connectorPushAttemptTestDDL = ` CREATE TABLE _System_ConnectorPushAttempt ( Id VARCHAR(255) NOT NULL, LinkKey VARCHAR(255) NOT NULL, AttemptOccupancyKey VARCHAR(255) NULL, Status VARCHAR(64) NOT NULL, DispatchLane VARCHAR(64) NOT NULL DEFAULT 'configured', AttemptOrdinal INT NOT NULL DEFAULT 1, RemoteId VARCHAR(512) NULL, ConnectorOwnerKey VARCHAR(255) NOT NULL DEFAULT '', ActionSlotId VARCHAR(255) NOT NULL DEFAULT '', ActionType VARCHAR(64) NOT NULL DEFAULT '', SourceObject VARCHAR(255) NOT NULL DEFAULT '', SourceRecordId VARCHAR(255) NOT NULL DEFAULT '', DestinationSlotId VARCHAR(255) NOT NULL DEFAULT '', RequesterOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ProfileOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ConnectorNamespace VARCHAR(255) NOT NULL DEFAULT '', FlowDeveloperName VARCHAR(255) NOT NULL DEFAULT '', FlowActionName VARCHAR(255) NOT NULL DEFAULT '', TargetObject VARCHAR(255) NOT NULL DEFAULT '', ProfileName VARCHAR(255) NOT NULL DEFAULT '', CreatedPackInstallId VARCHAR(255) NOT NULL DEFAULT '', CredentialRef VARCHAR(255) NOT NULL DEFAULT '', EndpointClass VARCHAR(255) NOT NULL DEFAULT '', RetryPolicy LONGTEXT NOT NULL, ProfileContractHash VARCHAR(255) NOT NULL DEFAULT '', CreatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', UpdatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', ResponseIdPath VARCHAR(255) NOT NULL DEFAULT '', RemoteIdField VARCHAR(255) NULL, RemoteIdProvenanceField VARCHAR(255) NULL, RemoteIdMirrorExpectedValue LONGTEXT NULL, RemoteIdMirrorCurrentValue LONGTEXT NULL, RemoteIdProvenanceMirrorExpectedValue LONGTEXT NULL, RemoteIdProvenanceMirrorCurrentValue LONGTEXT NULL, MappingFingerprint VARCHAR(255) NOT NULL DEFAULT '', BodyFingerprint VARCHAR(255) NOT NULL DEFAULT '', CreateConflictResolutionFingerprint VARCHAR(255) NOT NULL DEFAULT '', CreateConflictPatchIntentJSON JSON NULL, PatchDispatchIntentJSON JSON NULL, OriginatingUserId VARCHAR(255) NOT NULL DEFAULT '', OriginatingUserSessionJSON LONGTEXT NOT NULL, AllowedSourceFieldsJSON LONGTEXT NOT NULL, AssociationSourceRecordIdsJSON LONGTEXT NULL, RequestIdempotencyKey VARCHAR(255) NOT NULL DEFAULT '', RequestMethod VARCHAR(16) NOT NULL DEFAULT '', RequestRelativePath VARCHAR(1024) NOT NULL DEFAULT '', RequestHeaders LONGTEXT NOT NULL, RequestBodyBase64 LONGTEXT NOT NULL, RequestBodyHash VARCHAR(255) NOT NULL DEFAULT '', ResponseStatusCode INT NULL, -- nullable per the canonical schema (system_tables.json: ResponseStatusCode / -- ResponseHeaders / ResponseBodyBase64 are nullable:true). ResponseHeaders LONGTEXT NULL, ResponseBodyBase64 LONGTEXT NULL, ResponseBodyHash VARCHAR(255) NOT NULL DEFAULT '', CompatibilityMirrorStatus VARCHAR(64) NOT NULL DEFAULT '', RetryCount INT NOT NULL DEFAULT 0, LastError TEXT NULL, LastErrorAt DATETIME NULL, ClaimedBy VARCHAR(255) NULL, ClaimToken VARCHAR(255) NULL, ClaimedAt DATETIME NULL, LeaseExpiresAt DATETIME NULL, RemoteIOStartedAt DATETIME NULL, SupersededByAttemptId VARCHAR(255) NULL, SupersededAt DATETIME NULL, CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (Id), UNIQUE KEY uniq_connector_push_attempt_occupancy (AttemptOccupancyKey), UNIQUE KEY uniq_connector_push_attempt_ordinal (LinkKey, AttemptOrdinal), -- Full index set per the canonical schema (system_tables.json indices for -- _System_ConnectorPushAttempt), including the idempotency uniqueness guard so -- the create path is validated against a production-faithful surface. UNIQUE KEY uniq_connector_push_attempt_idempotency (RequestIdempotencyKey), KEY idx_connector_push_attempt_gate (LinkKey, Status), KEY idx_connector_push_attempt_claim (Status, ClaimedAt, CreatedDate), KEY idx_connector_push_attempt_source (SourceObject, SourceRecordId) ) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci` // TestSQLConnectorAdoptFinalizedLinkConsumesEvidenceAtomically proves that // AdoptFinalizedLink atomically creates the finalized link AND consumes the // authored evidence in a single transaction when EvidenceActiveKey is set. // Skipped when TiDB is unreachable. func TestSQLConnectorAdoptFinalizedLinkConsumesEvidenceAtomically(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() createConnectorRuntimeOwnerCatalogTables(t, db) ctx := context.Background() for _, ddl := range []string{connectorRemoteLinkTestDDL, connectorPushAttemptTestDDL, connectorInboundEvidenceTestDDL} { if _, err := db.ExecContext(ctx, ddl); err != nil { t.Fatalf("create table: %v", err) } } const ( owner = "owner-pack" ns = "connector_ns" obj = "Contact" rec = "003-atomic-adopt" field = "HubSpot_Id__c" ) store := NewSQLConnectorPushDurableStore(db) identity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: owner, ActionSlotID: "action-slot", ActionType: "PushToConnector", SourceObject: obj, SourceRecordID: rec, DestinationSlotID: "dest-slot", } linkKey, err := identity.LinkKey() if err != nil { t.Fatalf("LinkKey: %v", err) } snapshot := connectorPushTestSnapshot() snapshot.ConnectorNamespace = ns // --- 1. Seed an authored evidence row. --- const remoteID = "remote-adopt-test" activeKey := ConnectorInboundEvidenceActiveKey(owner, obj, rec, field) if err := store.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: "ev-adopt-1", ConnectorOwnerKey: owner, ConnectorNamespace: ns, SourceObject: obj, SourceRecordId: rec, RemoteIDField: field, RemoteId: remoteID, }); err != nil { t.Fatalf("seed authored evidence: %v", err) } // --- 2. AdoptFinalizedLink with matching EvidenceActiveKey. --- link, err := store.AdoptFinalizedLink(ctx, ConnectorPushLinkAdoption{ LinkKey: linkKey, RemoteID: remoteID, Identity: identity, Snapshot: snapshot, EvidenceActiveKey: activeKey, }) if err != nil { t.Fatalf("AdoptFinalizedLink: %v", err) } if link == nil || link.Status != ConnectorPushLinkStatusFinalized || link.RemoteID != remoteID { t.Fatalf("adopted link = %#v, want finalized/%s", link, remoteID) } // Verify: evidence row is consumed and ActiveKey is NULL. var evStatus, evConsumedBy, evActiveKey string if err := db.QueryRowContext(ctx, `SELECT EvidenceStatus, COALESCE(ConsumedByLinkKey,''), COALESCE(ActiveKey,'') FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?`, "ev-adopt-1").Scan(&evStatus, &evConsumedBy, &evActiveKey); err != nil { t.Fatalf("read evidence after adopt: %v", err) } if evStatus != ConnectorInboundEvidenceStatusConsumed { t.Fatalf("evidence status = %q, want consumed", evStatus) } if evConsumedBy != linkKey { t.Fatalf("evidence consumedByLinkKey = %q, want %q", evConsumedBy, linkKey) } if evActiveKey != "" { t.Fatalf("evidence ActiveKey = %q, want NULL/empty after consume", evActiveKey) } // --- 3. Idempotent second call returns the link without error. --- link2, err := store.AdoptFinalizedLink(ctx, ConnectorPushLinkAdoption{ LinkKey: linkKey, RemoteID: remoteID, Identity: identity, Snapshot: snapshot, EvidenceActiveKey: activeKey, // evidence already consumed; best-effort → no fault }) if err != nil { t.Fatalf("idempotent AdoptFinalizedLink: %v", err) } if link2 == nil || link2.Status != ConnectorPushLinkStatusFinalized { t.Fatalf("idempotent link = %#v, want finalized", link2) } // --- 4. Fail-closed: retire evidence, then attempt adopt with a fresh link key. --- // Seed a second authored evidence row for a different record to use for retirement. const rec2 = "003-atomic-adopt-2" activeKey2 := ConnectorInboundEvidenceActiveKey(owner, obj, rec2, field) if err := store.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{ ID: "ev-adopt-2", ConnectorOwnerKey: owner, ConnectorNamespace: ns, SourceObject: obj, SourceRecordId: rec2, RemoteIDField: field, RemoteId: remoteID, }); err != nil { t.Fatalf("seed evidence ev-adopt-2: %v", err) } // Retire it so it is no longer authored. if err := store.RetireEvidence(ctx, "ev-adopt-2"); err != nil { t.Fatalf("RetireEvidence: %v", err) } // Build a fresh identity for rec2 so the link key is different. identity2 := identity identity2.SourceRecordID = rec2 linkKey2, err := identity2.LinkKey() if err != nil { t.Fatalf("LinkKey2: %v", err) } snapshot2 := snapshot // AdoptFinalizedLink with the now-stale (retired) EvidenceActiveKey must fail // closed and leave no finalized link row. _, err = store.AdoptFinalizedLink(ctx, ConnectorPushLinkAdoption{ LinkKey: linkKey2, RemoteID: remoteID, Identity: identity2, Snapshot: snapshot2, EvidenceActiveKey: activeKey2, }) if !errors.Is(err, ErrConnectorPushDurableConflict) { t.Fatalf("stale evidence adopt error = %v, want ErrConnectorPushDurableConflict", err) } // Confirm no finalized link was created. var count int if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE LinkKey = ?", linkKey2).Scan(&count); err != nil { t.Fatalf("count link rows for linkKey2: %v", err) } if count != 0 { t.Fatalf("link row count for linkKey2 = %d, want 0 (rollback must have undone the insert)", count) } } // TestSQLConnectorInboundFinalizeCommitsOrRollsBackWithCaller is the real-DB // proof that the inbound primitive owns no transaction boundary. The caller's // rollback removes both rows; the caller's commit persists one finalized link // and one consumed evidence row without creating an outbound attempt. func TestSQLConnectorInboundFinalizeCommitsOrRollsBackWithCaller(t *testing.T) { db, cleanup := openTiDB(t) defer cleanup() createConnectorRuntimeOwnerCatalogTables(t, db) ctx := context.Background() for _, ddl := range []string{connectorRemoteLinkTestDDL, connectorPushAttemptTestDDL, connectorInboundEvidenceTestDDL} { if _, err := db.ExecContext(ctx, ddl); err != nil { t.Fatalf("create table: %v", err) } } if _, err := db.ExecContext(ctx, `CREATE TABLE _Test_InboundLocalRecord ( Id VARCHAR(255) NOT NULL, RemoteId VARCHAR(512) NOT NULL, PRIMARY KEY (Id) )`); err != nil { t.Fatalf("create local record sentinel table: %v", err) } identity := ConnectorPushDurableIdentity{ ConnectorOwnerKey: "owner-pack", ActionSlotID: "action-slot", ActionType: "PushToConnector", SourceObject: "Contact", SourceRecordID: "003-inbound-ambient", DestinationSlotID: "dest-slot", } snapshot := connectorPushTestSnapshot() snapshot.ConnectorNamespace = "connector_ns" snapshot.RemoteIDField = "HubSpot_Id__c" snapshot.ProfileContractHash = ConnectorPushProfileContractHash(snapshot) finalization := ConnectorInboundRemoteIdentityFinalization{ Evidence: ConnectorInboundRemoteIDEvidence{ ID: "ev-inbound-ambient", ConnectorOwnerKey: identity.ConnectorOwnerKey, ConnectorNamespace: "hubspot-routing-namespace", SourceObject: identity.SourceObject, SourceRecordId: identity.SourceRecordID, RemoteIDField: snapshot.RemoteIDField, RemoteId: "remote-inbound-ambient", }, Adoption: ConnectorPushLinkAdoption{ RemoteID: "remote-inbound-ambient", Identity: identity, Snapshot: snapshot, }, } linkKey := mustConnectorPushTestLinkKey(identity) countRows := func(table, where string, args ...any) int { t.Helper() var count int if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table+" WHERE "+where, args...).Scan(&count); err != nil { t.Fatalf("count %s: %v", table, err) } return count } txRollback, err := db.BeginTx(ctx, nil) if err != nil { t.Fatalf("begin rollback tx: %v", err) } rollbackStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txRollback}) if _, err := txRollback.ExecContext(ctx, "INSERT INTO _Test_InboundLocalRecord (Id, RemoteId) VALUES (?, ?)", identity.SourceRecordID, finalization.Adoption.RemoteID); err != nil { t.Fatalf("insert local record before rollback: %v", err) } if err := rollbackStore.InsertAuthoredEvidence(ctx, finalization.Evidence); err != nil { t.Fatalf("author evidence before rollback: %v", err) } if err := rollbackStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { t.Fatalf("finalize before rollback: %v", err) } if err := txRollback.Rollback(); err != nil { t.Fatalf("rollback: %v", err) } if got := countRows("_System_ConnectorRemoteLink", "LinkKey = ?", linkKey); got != 0 { t.Fatalf("link rows after caller rollback = %d, want 0", got) } if got := countRows("_System_ConnectorInboundRemoteIDEvidence", "Id = ?", finalization.Evidence.ID); got != 0 { t.Fatalf("evidence rows after caller rollback = %d, want 0", got) } if got := countRows("_Test_InboundLocalRecord", "Id = ?", identity.SourceRecordID); got != 0 { t.Fatalf("local rows after caller rollback = %d, want 0", got) } txCommit, err := db.BeginTx(ctx, nil) if err != nil { t.Fatalf("begin commit tx: %v", err) } commitStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txCommit}) if _, err := txCommit.ExecContext(ctx, "INSERT INTO _Test_InboundLocalRecord (Id, RemoteId) VALUES (?, ?)", identity.SourceRecordID, finalization.Adoption.RemoteID); err != nil { t.Fatalf("insert local record before commit: %v", err) } if err := commitStore.InsertAuthoredEvidence(ctx, finalization.Evidence); err != nil { t.Fatalf("author evidence before commit: %v", err) } if err := commitStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { t.Fatalf("finalize before commit: %v", err) } if err := txCommit.Commit(); err != nil { t.Fatalf("commit: %v", err) } var linkStatus, evidenceStatus, consumedBy string if err := db.QueryRowContext(ctx, "SELECT Status FROM _System_ConnectorRemoteLink WHERE LinkKey = ?", linkKey).Scan(&linkStatus); err != nil { t.Fatalf("read committed link: %v", err) } if err := db.QueryRowContext(ctx, "SELECT EvidenceStatus, COALESCE(ConsumedByLinkKey, '') FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = ?", finalization.Evidence.ID).Scan(&evidenceStatus, &consumedBy); err != nil { t.Fatalf("read committed evidence: %v", err) } if linkStatus != ConnectorPushLinkStatusFinalized || evidenceStatus != ConnectorInboundEvidenceStatusConsumed || consumedBy != linkKey { t.Fatalf("committed authority = link %q, evidence %q/%q; want finalized, consumed/%q", linkStatus, evidenceStatus, consumedBy, linkKey) } if got := countRows("_System_ConnectorPushAttempt", "LinkKey = ?", linkKey); got != 0 { t.Fatalf("attempt rows after inbound finalize = %d, want 0", got) } if got := countRows("_Test_InboundLocalRecord", "Id = ?", identity.SourceRecordID); got != 1 { t.Fatalf("local rows after caller commit = %d, want 1", got) } // A retry after the consumed-evidence commit is idempotent and performs no // new authoring because the identical finalized link is already authority. txRetry, err := db.BeginTx(ctx, nil) if err != nil { t.Fatalf("begin retry tx: %v", err) } retryStore := NewSQLConnectorPushDurableStoreWithTxExtractor(db, &testConnectorPushTxExtractor{tx: txRetry}) if err := retryStore.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil { t.Fatalf("idempotent retry: %v", err) } if err := txRetry.Commit(); err != nil { t.Fatalf("retry commit: %v", err) } if got := countRows("_System_ConnectorRemoteLink", "LinkKey = ?", linkKey); got != 1 { t.Fatalf("link rows after idempotent retry = %d, want 1", got) } if got := countRows("_System_ConnectorInboundRemoteIDEvidence", "ConsumedByLinkKey = ?", linkKey); got != 1 { t.Fatalf("consumed evidence rows after idempotent retry = %d, want 1", got) } } // --------------------------------------------------------------------------- // FindRecordPushState integration tests (#2107 - two-SSoT per-record read) // // These tests run against a live TiDB instance (TIDB_PORT=4000). They are // skipped when TiDB is unreachable. Each test uses its own isolated tables // via openSyncStatusTiDB which creates fresh tables per call. // --------------------------------------------------------------------------- // connectorSyncLinkDDL mirrors connectorRemoteLinkTestDDL but adds the // idx_connector_remote_link_source index used by Query A. const connectorSyncLinkDDL = ` CREATE TABLE _System_ConnectorRemoteLink ( Id VARCHAR(255) NOT NULL, LinkKey VARCHAR(255) NOT NULL, Status VARCHAR(64) NOT NULL, RemoteId VARCHAR(512) NOT NULL, RemoteLocatorHash VARCHAR(255) NOT NULL, RemoteLocatorActiveKey VARCHAR(255) NOT NULL, ConnectorOwnerKey VARCHAR(255) NOT NULL, ActionSlotId VARCHAR(255) NOT NULL, ActionType VARCHAR(64) NOT NULL, SourceObject VARCHAR(255) NOT NULL, SourceRecordId VARCHAR(255) NOT NULL, DestinationSlotId VARCHAR(255) NOT NULL, RequesterOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ProfileOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ConnectorNamespace VARCHAR(255) NOT NULL DEFAULT '', FlowDeveloperName VARCHAR(255) NOT NULL DEFAULT '', FlowActionName VARCHAR(255) NOT NULL DEFAULT '', TargetObject VARCHAR(255) NOT NULL DEFAULT '', ProfileName VARCHAR(255) NOT NULL DEFAULT '', CreatedPackInstallId VARCHAR(255) NOT NULL DEFAULT '', LastValidatedPackInstallId VARCHAR(255) NOT NULL DEFAULT '', CredentialRef VARCHAR(255) NOT NULL DEFAULT '', EndpointClass VARCHAR(255) NOT NULL DEFAULT '', RetryPolicy LONGTEXT NOT NULL, ProfileContractHash VARCHAR(255) NOT NULL DEFAULT '', CreatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', UpdatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', ResponseIdPath VARCHAR(255) NOT NULL DEFAULT '', RemoteIdField VARCHAR(255) NULL, RemoteIdProvenanceField VARCHAR(255) NULL, MappingFingerprint VARCHAR(255) NOT NULL DEFAULT '', BodyFingerprint VARCHAR(255) NOT NULL DEFAULT '', CompatibilityMirrorStatus VARCHAR(64) NOT NULL DEFAULT '', CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (Id), UNIQUE KEY uniq_connector_remote_link_key (LinkKey), UNIQUE KEY uniq_connector_remote_locator_active (RemoteLocatorActiveKey), KEY idx_connector_remote_link_source (SourceObject, SourceRecordId) )` // connectorSyncAttemptDDL mirrors connectorPushAttemptTestDDL but adds the // new idx_connector_push_attempt_source index (the schema addition of #2107). const connectorSyncAttemptDDL = ` CREATE TABLE _System_ConnectorPushAttempt ( Id VARCHAR(255) NOT NULL, LinkKey VARCHAR(255) NOT NULL, AttemptOccupancyKey VARCHAR(255) NULL, Status VARCHAR(64) NOT NULL, DispatchLane VARCHAR(64) NOT NULL DEFAULT 'configured', AttemptOrdinal INT NOT NULL DEFAULT 1, RemoteId VARCHAR(512) NULL, ConnectorOwnerKey VARCHAR(255) NOT NULL DEFAULT '', ActionSlotId VARCHAR(255) NOT NULL DEFAULT '', ActionType VARCHAR(64) NOT NULL DEFAULT '', SourceObject VARCHAR(255) NOT NULL DEFAULT '', SourceRecordId VARCHAR(255) NOT NULL DEFAULT '', DestinationSlotId VARCHAR(255) NOT NULL DEFAULT '', RequesterOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ProfileOwnerPack VARCHAR(255) NOT NULL DEFAULT '', ConnectorNamespace VARCHAR(255) NOT NULL DEFAULT '', FlowDeveloperName VARCHAR(255) NOT NULL DEFAULT '', FlowActionName VARCHAR(255) NOT NULL DEFAULT '', TargetObject VARCHAR(255) NOT NULL DEFAULT '', ProfileName VARCHAR(255) NOT NULL DEFAULT '', CreatedPackInstallId VARCHAR(255) NOT NULL DEFAULT '', CredentialRef VARCHAR(255) NOT NULL DEFAULT '', EndpointClass VARCHAR(255) NOT NULL DEFAULT '', RetryPolicy LONGTEXT NOT NULL, ProfileContractHash VARCHAR(255) NOT NULL DEFAULT '', CreatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', UpdatePathTemplate VARCHAR(1024) NOT NULL DEFAULT '', ResponseIdPath VARCHAR(255) NOT NULL DEFAULT '', RemoteIdField VARCHAR(255) NULL, RemoteIdProvenanceField VARCHAR(255) NULL, RemoteIdMirrorExpectedValue LONGTEXT NULL, RemoteIdMirrorCurrentValue LONGTEXT NULL, RemoteIdProvenanceMirrorExpectedValue LONGTEXT NULL, RemoteIdProvenanceMirrorCurrentValue LONGTEXT NULL, MappingFingerprint VARCHAR(255) NOT NULL DEFAULT '', BodyFingerprint VARCHAR(255) NOT NULL DEFAULT '', PatchDispatchIntentJSON JSON NULL, OriginatingUserId VARCHAR(255) NOT NULL DEFAULT '', OriginatingUserSessionJSON LONGTEXT NOT NULL, AllowedSourceFieldsJSON LONGTEXT NOT NULL, AssociationSourceRecordIdsJSON LONGTEXT NULL, RequestIdempotencyKey VARCHAR(255) NOT NULL DEFAULT '', RequestMethod VARCHAR(16) NOT NULL DEFAULT '', RequestRelativePath VARCHAR(1024) NOT NULL DEFAULT '', RequestHeaders LONGTEXT NOT NULL, RequestBodyBase64 LONGTEXT NOT NULL, RequestBodyHash VARCHAR(255) NOT NULL DEFAULT '', ResponseStatusCode INT NOT NULL DEFAULT 0, ResponseHeaders LONGTEXT NOT NULL, ResponseBodyBase64 LONGTEXT NOT NULL, ResponseBodyHash VARCHAR(255) NOT NULL DEFAULT '', CompatibilityMirrorStatus VARCHAR(64) NOT NULL DEFAULT '', RetryCount INT NOT NULL DEFAULT 0, LastError TEXT NULL, LastErrorAt DATETIME NULL, ClaimedBy VARCHAR(255) NULL, ClaimToken VARCHAR(255) NULL, ClaimedAt DATETIME NULL, LeaseExpiresAt DATETIME NULL, RemoteIOStartedAt DATETIME NULL, SupersededByAttemptId VARCHAR(255) NULL, SupersededAt DATETIME NULL, CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (Id), UNIQUE KEY uniq_connector_push_attempt_occupancy (AttemptOccupancyKey), UNIQUE KEY uniq_connector_push_attempt_ordinal (LinkKey, AttemptOrdinal), KEY idx_connector_push_attempt_gate (LinkKey, Status), KEY idx_connector_push_attempt_claim (Status, ClaimedAt, CreatedDate), KEY idx_connector_push_attempt_source (SourceObject, SourceRecordId) )` // openSyncStatusTiDB opens a TiDB database with both tables and the new // source index created. Availability follows openTiDB's configured endpoint // and fail-closed required-mode contract. func openSyncStatusTiDB(t *testing.T) (*sql.DB, func()) { t.Helper() db, cleanup := openTiDB(t) t.Cleanup(func() { cleanup() }) ctx := t.Context() for _, ddl := range []string{connectorSyncLinkDDL, connectorSyncAttemptDDL} { if _, err := db.ExecContext(ctx, ddl); err != nil { t.Fatalf("create sync-status table: %v", err) } } return db, cleanup } // insertSyncLink inserts a minimal _System_ConnectorRemoteLink row. func insertSyncLink(t *testing.T, db *sql.DB, id, linkKey, status, remoteId, ownerKey, actionSlot, actionType, sourceObj, sourceRec, destSlot, targetObj string) { t.Helper() _, err := db.ExecContext(context.Background(), ` INSERT INTO _System_ConnectorRemoteLink (Id, LinkKey, Status, RemoteId, RemoteLocatorHash, RemoteLocatorActiveKey, ConnectorOwnerKey, ActionSlotId, ActionType, SourceObject, SourceRecordId, DestinationSlotId, TargetObject, RetryPolicy, CompatibilityMirrorStatus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '{"MaxAttempts":1}', 'synced')`, id, linkKey, status, remoteId, "hash-"+id, "hash-"+id, ownerKey, actionSlot, actionType, sourceObj, sourceRec, destSlot, targetObj, ) if err != nil { t.Fatalf("insertSyncLink %s: %v", id, err) } } // insertSyncAttempt inserts a minimal _System_ConnectorPushAttempt row. func insertSyncAttempt(t *testing.T, db *sql.DB, id, linkKey, status string, ordinal int, statusCode int, lastError string, ownerKey, actionSlot, actionType, sourceObj, sourceRec, destSlot, targetObj string) { t.Helper() lastErrorExpr := "NULL" lastErrorAtExpr := "NULL" if lastError != "" { lastErrorExpr = "'" + strings.ReplaceAll(lastError, "'", "\\'") + "'" lastErrorAtExpr = "CURRENT_TIMESTAMP" } query := `INSERT INTO _System_ConnectorPushAttempt (Id, LinkKey, Status, AttemptOrdinal, ResponseStatusCode, ConnectorOwnerKey, ActionSlotId, ActionType, SourceObject, SourceRecordId, DestinationSlotId, TargetObject, RetryPolicy, ProfileContractHash, MappingFingerprint, BodyFingerprint, OriginatingUserId, OriginatingUserSessionJSON, AllowedSourceFieldsJSON, RequestIdempotencyKey, RequestMethod, RequestRelativePath, RequestHeaders, RequestBodyBase64, RequestBodyHash, ResponseHeaders, ResponseBodyBase64, ResponseBodyHash, LastError, LastErrorAt) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '{"MaxAttempts":1}', 'ph', 'mfp', 'bfp', '005', '{"ID":"005"}', '[]', CONCAT('idem-', ?), 'POST', '/objects', '{}', '', '', '{}', '', '', ` + lastErrorExpr + `, ` + lastErrorAtExpr + `)` _, err := db.ExecContext(context.Background(), query, id, linkKey, status, ordinal, statusCode, ownerKey, actionSlot, actionType, sourceObj, sourceRec, destSlot, targetObj, id, ) if err != nil { t.Fatalf("insertSyncAttempt %s: %v", id, err) } } // T1: Action-slot rename collapse — two attempts with different LinkKeys (same // DestinationSlotId, same ConnectorOwnerKey, same ActionType) collapse to ONE row. func TestFindRecordPushState_T1_ActionSlotRenameCollapse(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t1" owner = "owner-t1" slot1 = "slot-t1-old" // old action slot slot2 = "slot-t1-new" // new action slot (rename) dest = "dest-hubspot" // same DestinationSlotId for both target = "contacts" atype = "PushToConnector" ) // Two LinkKeys for the same 5-tuple (rename scenario). // The 5-tuple key = (ConnectorOwnerKey=owner, ActionType=atype, DestinationSlotId=dest). // Link for new slot (higher LastModifiedDate). insertSyncLink(t, db, "link-t1-old", "lk-t1-old", ConnectorPushLinkStatusFinalized, "remote-old", owner, slot1, atype, obj, rec, dest, target) insertSyncLink(t, db, "link-t1-new", "lk-t1-new", ConnectorPushLinkStatusFinalized, "remote-new", owner, slot2, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T1: %v", err) } if len(results) != 1 { t.Fatalf("T1: expected 1 row (action-slot rename collapses to 1 per 5-tuple); got %d: %+v", len(results), results) } r := results[0] if !r.HasLink { t.Errorf("T1: HasLink=false, want true") } if r.DestinationSlotID != dest { t.Errorf("T1: DestinationSlotID=%q, want %q", r.DestinationSlotID, dest) } } // T2: Two connectors with the same DestinationSlotId → two rows (F1). // Synthesized: no live instances of this scenario. func TestFindRecordPushState_T2_TwoConnectorsSameDestination(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t2" ownerA = "owner-t2-a" ownerB = "owner-t2-b" slot = "slot-t2" dest = "dest-shared" // SAME destination slot id target = "contacts" atype = "PushToConnector" ) insertSyncLink(t, db, "link-t2-a", "lk-t2-a", ConnectorPushLinkStatusFinalized, "remote-a", ownerA, slot, atype, obj, rec, dest, target) insertSyncLink(t, db, "link-t2-b", "lk-t2-b", ConnectorPushLinkStatusFinalized, "remote-b", ownerB, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T2: %v", err) } if len(results) != 2 { t.Fatalf("T2: expected 2 rows (two connectors same dest); got %d: %+v", len(results), results) } // Both rows must have HasLink=true. for _, r := range results { if !r.HasLink { t.Errorf("T2: row HasLink=false: %+v", r) } } // OQ-2: the two rows must be distinguishable by ConnectorOwnerKey. // With only DestinationSlotID+TargetObject, the rows would be identical; // ConnectorOwnerKey is the lineage discriminator. if results[0].ConnectorOwnerKey == results[1].ConnectorOwnerKey { t.Errorf("T2: both rows have the same ConnectorOwnerKey %q; expected %q and %q", results[0].ConnectorOwnerKey, ownerA, ownerB) } owners := map[string]bool{results[0].ConnectorOwnerKey: true, results[1].ConnectorOwnerKey: true} if !owners[ownerA] || !owners[ownerB] { t.Errorf("T2: expected ConnectorOwnerKey values {%q, %q}; got {%q, %q}", ownerA, ownerB, results[0].ConnectorOwnerKey, results[1].ConnectorOwnerKey) } } // T3: Two different DestinationSlotIds → two rows disambiguated by slot/target. func TestFindRecordPushState_T3_TwoDestinations(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t3" owner = "owner-t3" slot = "slot-t3" dest1 = "dest-t3-company" dest2 = "dest-t3-contact" tgt1 = "companies" tgt2 = "contacts" atype = "PushToConnector" ) insertSyncLink(t, db, "link-t3-1", "lk-t3-1", ConnectorPushLinkStatusFinalized, "remote-1", owner, slot, atype, obj, rec, dest1, tgt1) insertSyncLink(t, db, "link-t3-2", "lk-t3-2", ConnectorPushLinkStatusFinalized, "remote-2", owner, slot, atype, obj, rec, dest2, tgt2) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T3: %v", err) } if len(results) != 2 { t.Fatalf("T3: expected 2 rows (two destinations); got %d: %+v", len(results), results) } // Results must be sorted by (DestinationSlotId, ConnectorOwnerKey). if results[0].DestinationSlotID > results[1].DestinationSlotID { t.Errorf("T3: results not sorted by DestinationSlotID: %q > %q", results[0].DestinationSlotID, results[1].DestinationSlotID) } destSet := map[string]bool{} for _, r := range results { destSet[r.DestinationSlotID] = true } if !destSet[dest1] || !destSet[dest2] { t.Errorf("T3: expected both destinations in results; got %v", destSet) } } // T4: Drift/retire passthrough — link is authoritative for sync state. // suspended_drift and retired links must appear with link fields set. // T4: suspended_drift is surfaced; retired is excluded — mirrors FindStateByDestination. // The retired destination (dest-t4-retired) has no attempt, so after the link // filter excludes its link row there is nothing to emit for that 5-tuple. func TestFindRecordPushState_T4_DriftSurfacedRetiredExcluded(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Account" rec = "001-t4" owner = "owner-t4" slot = "slot-t4" atype = "PushToConnector" ) insertSyncLink(t, db, "link-t4-drift", "lk-t4-drift", ConnectorPushLinkStatusSuspendedDrift, "remote-drift", owner, slot+"-drift", atype, obj, rec, "dest-t4-drift", "companies") insertSyncLink(t, db, "link-t4-retired", "lk-t4-retired", ConnectorPushLinkStatusRetired, "remote-retired", owner, slot+"-retired", atype, obj, rec, "dest-t4-retired", "deals") store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T4: %v", err) } // Only the suspended_drift row is returned; retired is excluded by the status // filter, and that destination has no attempt row to pull it in. if len(results) != 1 { t.Fatalf("T4: expected 1 row; got %d: %+v", len(results), results) } r := results[0] if !r.HasLink { t.Errorf("T4: expected HasLink=true; got %+v", r) } if r.LinkStatus != ConnectorPushLinkStatusSuspendedDrift { t.Errorf("T4: expected LinkStatus=%q; got %q", ConnectorPushLinkStatusSuspendedDrift, r.LinkStatus) } if r.DestinationSlotID != "dest-t4-drift" { t.Errorf("T4: expected DestinationSlotID=%q; got %q", "dest-t4-drift", r.DestinationSlotID) } // retired destination must not appear for _, row := range results { if row.DestinationSlotID == "dest-t4-retired" { t.Errorf("T4: retired destination unexpectedly in results: %+v", row) } if row.LinkStatus == ConnectorPushLinkStatusRetired { t.Errorf("T4: retired link status unexpectedly in results: %+v", row) } } } // TestFindRecordPushState_RetiredDoesNotShadowFinalized is the v15 shadowing // regression guard: for one destination 5-tuple insert a NEWER retired link and // an OLDER finalized link. On the pre-fix query (no status filter) the retired // link wins LastModifiedDate-DESC collapse and the row reports "retired". With // the fix, the retired link is excluded and the finalized row is returned. func TestFindRecordPushState_RetiredDoesNotShadowFinalized(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Account" rec = "001-shadow" owner = "owner-shadow" slot = "slot-shadow" atype = "PushToConnector" dest = "dest-shadow" ) // Insert finalized link first (will get an earlier CURRENT_TIMESTAMP). insertSyncLink(t, db, "link-shadow-finalized", "lk-shadow-fin", ConnectorPushLinkStatusFinalized, "remote-fin", owner, slot+"-fin", atype, obj, rec, dest, "companies") // Backdate the finalized link to make the ordering unambiguous. if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET LastModifiedDate = '2020-01-01 00:00:00' WHERE Id = 'link-shadow-finalized'`); err != nil { t.Fatalf("backdate finalized: %v", err) } // Insert retired link with a strictly NEWER LastModifiedDate. insertSyncLink(t, db, "link-shadow-retired", "lk-shadow-ret", ConnectorPushLinkStatusRetired, "remote-ret", owner, slot+"-ret", atype, obj, rec, dest, "companies") if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET LastModifiedDate = '2030-01-01 00:00:00' WHERE Id = 'link-shadow-retired'`); err != nil { t.Fatalf("bump retired: %v", err) } store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("shadow: %v", err) } // Retired link excluded by status filter → only finalized survives. if len(results) != 1 { t.Fatalf("shadow: expected 1 row; got %d: %+v", len(results), results) } r := results[0] if !r.HasLink { t.Errorf("shadow: expected HasLink=true; got %+v", r) } if r.LinkStatus != ConnectorPushLinkStatusFinalized { t.Errorf("shadow: expected LinkStatus=%q; got %q (retired newer link must not shadow finalized)", ConnectorPushLinkStatusFinalized, r.LinkStatus) } } // T5: Adopted link with no attempt → HasAttempt=false (lastAttemptStatus null). func TestFindRecordPushState_T5_AdoptedLinkNoAttempt(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Contact" rec = "003-t5" owner = "owner-t5" slot = "slot-t5" dest = "dest-t5" target = "contacts" linkKey = "lk-t5" atype = "PushToConnector" ) insertSyncLink(t, db, "link-t5", linkKey, ConnectorPushLinkStatusFinalized, "remote-t5", owner, slot, atype, obj, rec, dest, target) // No attempt row. store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T5: %v", err) } if len(results) != 1 { t.Fatalf("T5: expected 1 result; got %d: %+v", len(results), results) } r := results[0] if !r.HasLink { t.Errorf("T5: HasLink=false, want true") } if r.LinkStatus != ConnectorPushLinkStatusFinalized { t.Errorf("T5: LinkStatus=%q, want finalized", r.LinkStatus) } if r.HasAttempt { t.Errorf("T5: HasAttempt=true, want false (no attempt row)") } if r.LastAttemptStatus != "" { t.Errorf("T5: LastAttemptStatus=%q, want empty (no attempt)", r.LastAttemptStatus) } } // T6: Failed-never-linked — LinkStatus null, LastAttemptStatus=terminal_failed. func TestFindRecordPushState_T6_FailedNeverLinked(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Opportunity" rec = "006-t6" owner = "owner-t6" slot = "slot-t6" dest = "dest-t6" target = "deals" linkKey = "lk-t6" atype = "PushToConnector" ) // No link row — only a terminal_failed attempt. insertSyncAttempt(t, db, "att-t6", linkKey, ConnectorPushAttemptStatusTerminalFailed, 1, 422, "connection refused", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T6: %v", err) } if len(results) != 1 { t.Fatalf("T6: expected 1 result; got %d: %+v", len(results), results) } r := results[0] if r.HasLink { t.Errorf("T6: HasLink=true, want false (no link row)") } if r.LinkStatus != "" { t.Errorf("T6: LinkStatus=%q, want empty", r.LinkStatus) } if !r.HasAttempt { t.Errorf("T6: HasAttempt=false, want true") } if r.LastAttemptStatus != ConnectorPushAttemptStatusTerminalFailed { t.Errorf("T6: LastAttemptStatus=%q, want terminal_failed", r.LastAttemptStatus) } } // T6b: ambiguous_remote_effect with NULL responseStatusCode → status string // surfaced, no raw LastError text in the result struct. func TestFindRecordPushState_T6b_AmbiguousRemoteEffectNullCode(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t6b" owner = "owner-t6b" slot = "slot-t6b" dest = "dest-t6b" target = "contacts" linkKey = "lk-t6b" atype = "PushToConnector" ) // Attempt with statusCode=0 (will be NULL in DB) and raw LastError. insertSyncAttempt(t, db, "att-t6b", linkKey, ConnectorPushAttemptStatusAmbiguousRemoteEffect, 1, 0, "remote timeout — status unknown", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T6b: %v", err) } if len(results) != 1 { t.Fatalf("T6b: expected 1 result; got %d: %+v", len(results), results) } r := results[0] if r.LastAttemptStatus != ConnectorPushAttemptStatusAmbiguousRemoteEffect { t.Errorf("T6b: LastAttemptStatus=%q, want ambiguous_remote_effect", r.LastAttemptStatus) } // ResponseStatusCode is sql.NullInt64; statusCode=0 was stored as 0 (not NULL) // but the design allows NULL — verify the status string is the signal. // The DTO must NOT carry raw LastError. // (LastError is not a field of ConnectorPushRecordDestinationState — structural guarantee.) _ = r.ResponseStatusCode // valid check: no panic // Verify HasAttempt is true. if !r.HasAttempt { t.Errorf("T6b: HasAttempt=false, want true") } } // T15: EXPLAIN proves Query B rides idx_connector_push_attempt_source. func TestFindRecordPushState_T15_ExplainUsesAttemptSourceIndex(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() // Use a unique prefix so this test doesn't clash with others. const q = `EXPLAIN SELECT ConnectorOwnerKey, ActionType, DestinationSlotId, TargetObject, Status, ResponseStatusCode, LastErrorAt, LastModifiedDate, LinkKey FROM _System_ConnectorPushAttempt WHERE SourceObject = ? AND SourceRecordId = ? AND RequestMethod = 'POST' AND Status <> 'superseded'` rows, err := db.QueryContext(context.Background(), q, "Lead", "00Q-explain-test") if err != nil { t.Fatalf("T15: EXPLAIN query error: %v", err) } defer rows.Close() cols, _ := rows.Columns() var planLines []string for rows.Next() { vals := make([]interface{}, len(cols)) ptrs := make([]interface{}, len(cols)) for i := range vals { ptrs[i] = &vals[i] } if err := rows.Scan(ptrs...); err != nil { t.Fatalf("T15: EXPLAIN scan: %v", err) } parts := make([]string, len(cols)) for i, v := range vals { if v == nil { parts[i] = "" } else { switch sv := v.(type) { case []byte: parts[i] = string(sv) default: parts[i] = fmt.Sprintf("%v", sv) } } } planLines = append(planLines, strings.Join(parts, " | ")) } if err := rows.Err(); err != nil { t.Fatalf("T15: EXPLAIN rows err: %v", err) } planStr := strings.Join(planLines, "\n") if !strings.Contains(planStr, "idx_connector_push_attempt_source") { t.Errorf("T15: EXPLAIN output does not reference idx_connector_push_attempt_source; plan:\n%s", planStr) } t.Logf("T15 EXPLAIN plan:\n%s", planStr) } // T16: Superseded excluded / finalized+terminal_failed kept. func TestFindRecordPushState_T16_SupersededExcluded_FinalizedKept(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t16" owner = "owner-t16" slot = "slot-t16" dest = "dest-t16" target = "contacts" linkKey = "lk-t16" atype = "PushToConnector" ) // ordinal 1: finalized (KEEP) insertSyncAttempt(t, db, "att-t16-1", linkKey, ConnectorPushAttemptStatusFinalized, 1, 200, "", owner, slot, atype, obj, rec, dest, target) // ordinal 2: terminal_failed (KEEP) insertSyncAttempt(t, db, "att-t16-2", linkKey, ConnectorPushAttemptStatusTerminalFailed, 2, 422, "rejected", owner, slot, atype, obj, rec, dest, target) // ordinal 3: superseded (EXCLUDE) insertSyncAttempt(t, db, "att-t16-3", linkKey, ConnectorPushAttemptStatusSuperseded, 3, 0, "", owner, slot, atype, obj, rec, dest, target) // Also add a link. insertSyncLink(t, db, "link-t16", linkKey, ConnectorPushLinkStatusFinalized, "remote-t16", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T16: %v", err) } if len(results) != 1 { t.Fatalf("T16: expected 1 result; got %d: %+v", len(results), results) } r := results[0] // The latest non-superseded attempt is ordinal 2 (terminal_failed). if r.LastAttemptStatus != ConnectorPushAttemptStatusTerminalFailed { t.Errorf("T16: LastAttemptStatus=%q, want terminal_failed (superseded ordinal 3 excluded)", r.LastAttemptStatus) } } // T17: Tiebreaker determinism — attempt side and link side. // // Attempt side (T17a): a superseded attempt (Status='superseded') is excluded // by the SQL predicate; only the non-superseded attempt is surfaced. // Then two non-superseded attempts with the same time → higher Id wins (T17c). // // Link side (T17b, P1-2 regression): an OLDER finalized link vs. a NEWER // suspended_drift link for the same 5-tuple. The NEWER row must win // (LastModifiedDate DESC), not the finalized one. A stale finalized link // shadowing a newer drifted/retired link is the exact P1-2 bug. // // Link side (T17b2): same-LastModifiedDate link tiebreaker — higher Id wins. func TestFindRecordPushState_T17_TiebreakerDeterminism(t *testing.T) { t.Run("attempt_superseded_excluded", func(t *testing.T) { // T17a: superseded attempts are filtered by SQL (Status <> 'superseded'). // Only the non-superseded attempt should appear. db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t17a" owner = "owner-t17a" slot = "slot-t17a" dest = "dest-t17a" target = "contacts" atype = "PushToConnector" ) // Insert a superseded attempt (Status='superseded') — must be excluded by SQL. insertSyncAttempt(t, db, "att-t17a-sup", "lk-t17a", ConnectorPushAttemptStatusSuperseded, 1, 422, "err-sup", owner, slot, atype, obj, rec, dest, target) // Insert the current (non-superseded) attempt. insertSyncAttempt(t, db, "att-t17a-cur", "lk-t17a", ConnectorPushAttemptStatusAmbiguousRemoteEffect, 2, 0, "err-cur", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17a: %v", err) } if len(results) != 1 { t.Fatalf("T17a: expected 1 result; got %d: %+v", len(results), results) } // Only the non-superseded attempt must appear. if results[0].LastAttemptStatus != ConnectorPushAttemptStatusAmbiguousRemoteEffect { t.Errorf("T17a: LastAttemptStatus=%q, want ambiguous_remote_effect (superseded excluded by SQL predicate)", results[0].LastAttemptStatus) } // Run twice — determinism check. results2, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17a round2: %v", err) } if len(results2) != 1 || results2[0].LastAttemptStatus != results[0].LastAttemptStatus { t.Errorf("T17a: non-deterministic: round1=%q round2=%q", results[0].LastAttemptStatus, results2[0].LastAttemptStatus) } }) t.Run("attempt_id_tiebreaker", func(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t17c" owner = "owner-t17c" slot = "slot-t17c" dest = "dest-t17c" target = "contacts" atype = "PushToConnector" ) // Two non-superseded attempts with different Ids but same LastModifiedDate. // Higher Id ("zzz-t17c") wins. insertSyncAttempt(t, db, "aaa-t17c", "lk-t17c", ConnectorPushAttemptStatusTerminalFailed, 1, 422, "err-a", owner, slot, atype, obj, rec, dest, target) insertSyncAttempt(t, db, "zzz-t17c", "lk-t17c", ConnectorPushAttemptStatusAmbiguousRemoteEffect, 2, 0, "err-z", owner, slot, atype, obj, rec, dest, target) if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorPushAttempt SET LastModifiedDate = '2026-01-01 00:00:00' WHERE Id IN ('aaa-t17c', 'zzz-t17c')`); err != nil { t.Fatalf("T17c: force time: %v", err) } store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17c: %v", err) } if len(results) != 1 { t.Fatalf("T17c: expected 1 result; got %d", len(results)) } if results[0].LastAttemptStatus != ConnectorPushAttemptStatusAmbiguousRemoteEffect { t.Errorf("T17c: LastAttemptStatus=%q, want ambiguous_remote_effect (zzz > aaa wins Id tiebreaker)", results[0].LastAttemptStatus) } // Run twice — determinism check. results2, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17c round2: %v", err) } if len(results2) != 1 || results2[0].LastAttemptStatus != results[0].LastAttemptStatus { t.Errorf("T17c: non-deterministic: round1=%q round2=%q", results[0].LastAttemptStatus, results2[0].LastAttemptStatus) } }) t.Run("link_newer_beats_older", func(t *testing.T) { // T17b — P1-2 regression: a NEWER suspended_drift/retired link must beat // an OLDER finalized link for the same 5-tuple. // The old status-currency comparator picked finalized regardless of timestamp, // causing a stale finalized link to shadow a current drifted/retired state. // The fix uses LastModifiedDate DESC as primary sort, so the newer row wins. db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t17b" owner = "owner-t17b" slot = "slot-t17b" dest = "dest-t17b" target = "contacts" atype = "PushToConnector" ) // Older finalized link (timestamp 2026-01-01). insertSyncLink(t, db, "link-t17b-final", "lk-t17b-fin", ConnectorPushLinkStatusFinalized, "remote-fin", owner, slot, atype, obj, rec, dest, target) if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET LastModifiedDate = '2026-01-01 00:00:00' WHERE Id = 'link-t17b-final'`); err != nil { t.Fatalf("T17b: set old time: %v", err) } // Newer suspended_drift link (timestamp 2026-06-01) — must win. insertSyncLink(t, db, "link-t17b-drift", "lk-t17b-dft", ConnectorPushLinkStatusSuspendedDrift, "remote-dft", owner, slot, atype, obj, rec, dest, target) if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET LastModifiedDate = '2026-06-01 00:00:00' WHERE Id = 'link-t17b-drift'`); err != nil { t.Fatalf("T17b: set new time: %v", err) } store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17b: %v", err) } if len(results) != 1 { t.Fatalf("T17b: expected 1 result; got %d: %+v", len(results), results) } // Newer suspended_drift must win; stale finalized must NOT shadow it. if results[0].LinkStatus != ConnectorPushLinkStatusSuspendedDrift { t.Errorf("T17b: LinkStatus=%q, want suspended_drift (newer LastModifiedDate beats older finalized — P1-2 regression)", results[0].LinkStatus) } // Run twice — determinism check. results2, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17b round2: %v", err) } if len(results2) != 1 || results2[0].LinkStatus != results[0].LinkStatus { t.Errorf("T17b: non-deterministic: round1=%q round2=%q", results[0].LinkStatus, results2[0].LinkStatus) } }) t.Run("link_id_tiebreaker", func(t *testing.T) { // T17b2: same LastModifiedDate for two links — higher Id wins. // Both links are within the status filter (finalized / suspended_drift) // so this exercises the pure Id tiebreaker path. retired is excluded by // the status filter and is therefore not a valid Id-tiebreaker candidate. db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t17b2" owner = "owner-t17b2" slot = "slot-t17b2" dest = "dest-t17b2" target = "contacts" atype = "PushToConnector" ) // "aaa-link" (finalized) and "zzz-link" (suspended_drift) — same time. // "zzz-link" > "aaa-link" lexicographically, so suspended_drift wins. insertSyncLink(t, db, "aaa-link", "lk-b2-aaa", ConnectorPushLinkStatusFinalized, "remote-aaa", owner, slot, atype, obj, rec, dest, target) insertSyncLink(t, db, "zzz-link", "lk-b2-zzz", ConnectorPushLinkStatusSuspendedDrift, "remote-zzz", owner, slot, atype, obj, rec, dest, target) if _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorRemoteLink SET LastModifiedDate = '2026-01-01 00:00:00' WHERE Id IN ('aaa-link', 'zzz-link')`); err != nil { t.Fatalf("T17b2: force time: %v", err) } store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17b2: %v", err) } if len(results) != 1 { t.Fatalf("T17b2: expected 1 result; got %d: %+v", len(results), results) } // "zzz-link" > "aaa-link" — suspended_drift wins Id tiebreaker. if results[0].LinkStatus != ConnectorPushLinkStatusSuspendedDrift { t.Errorf("T17b2: LinkStatus=%q, want suspended_drift (zzz > aaa wins Id tiebreaker)", results[0].LinkStatus) } // Run twice — determinism check. results2, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T17b2 round2: %v", err) } if len(results2) != 1 || results2[0].LinkStatus != results[0].LinkStatus { t.Errorf("T17b2: non-deterministic: round1=%q round2=%q", results[0].LinkStatus, results2[0].LinkStatus) } }) } // insertSyncAttemptWithMethod is like insertSyncAttempt but lets the caller // specify RequestMethod (e.g. "PATCH") for PATCH-path regression tests. func insertSyncAttemptWithMethod(t *testing.T, db *sql.DB, id, linkKey, status, requestMethod string, ordinal int, statusCode int, lastError string, ownerKey, actionSlot, actionType, sourceObj, sourceRec, destSlot, targetObj string) { t.Helper() lastErrorExpr := "NULL" lastErrorAtExpr := "NULL" if lastError != "" { lastErrorExpr = "'" + strings.ReplaceAll(lastError, "'", "\\'") + "'" lastErrorAtExpr = "CURRENT_TIMESTAMP" } query := `INSERT INTO _System_ConnectorPushAttempt (Id, LinkKey, Status, AttemptOrdinal, ResponseStatusCode, ConnectorOwnerKey, ActionSlotId, ActionType, SourceObject, SourceRecordId, DestinationSlotId, TargetObject, RetryPolicy, ProfileContractHash, MappingFingerprint, BodyFingerprint, OriginatingUserId, OriginatingUserSessionJSON, AllowedSourceFieldsJSON, RequestIdempotencyKey, RequestMethod, RequestRelativePath, RequestHeaders, RequestBodyBase64, RequestBodyHash, ResponseHeaders, ResponseBodyBase64, ResponseBodyHash, LastError, LastErrorAt) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '{"MaxAttempts":1}', 'ph', 'mfp', 'bfp', '005', '{"ID":"005"}', '[]', CONCAT('idem-', ?), ?, '/objects', '{}', '', '', '{}', '', '', ` + lastErrorExpr + `, ` + lastErrorAtExpr + `)` _, err := db.ExecContext(context.Background(), query, id, linkKey, status, ordinal, statusCode, ownerKey, actionSlot, actionType, sourceObj, sourceRec, destSlot, targetObj, id, requestMethod, ) if err != nil { t.Fatalf("insertSyncAttemptWithMethod %s: %v", id, err) } } // T18: PATCH terminal_failed surfaced alongside finalized link. // // Regression for the AND RequestMethod='POST' predicate that was previously // in findRecordPushStateQueryAttempt: once a record has a finalized link the // normal sync path records PATCH attempts; a PATCH terminal_failed was // invisible to FindRecordPushState, making /admin/connector/sync-status show // a stale "synced" view after an update failure. // // This test seeds: // - one _System_ConnectorRemoteLink row (Status=finalized) // - one POST attempt (Status=finalized, older LastModifiedDate) — the original create // - one PATCH attempt (Status=terminal_failed, newer LastModifiedDate) // // Before the fix: FindRecordPushState returned HasAttempt=true with // LastAttemptStatus=finalized (POST) and missed the PATCH failure. // After the fix: LastAttemptStatus=terminal_failed (PATCH wins latest-wins). func TestFindRecordPushState_T18_PatchTerminalFailedSurfaced(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-t18" owner = "owner-t18" slot = "slot-t18" dest = "dest-t18" target = "contacts" atype = "PushToConnector" linkKey = "lk-t18" ) // Link row: finalized (record was created successfully via POST). insertSyncLink(t, db, "link-t18", linkKey, ConnectorPushLinkStatusFinalized, "remote-t18", owner, slot, atype, obj, rec, dest, target) // POST attempt (ordinal 1): finalized — the successful create, older timestamp. insertSyncAttemptWithMethod(t, db, "attempt-t18-post", linkKey, ConnectorPushAttemptStatusFinalized, "POST", 1, 200, "", owner, slot, atype, obj, rec, dest, target) // PATCH attempt (ordinal 2): terminal_failed — the update failure, newer timestamp. // Sleep 1s to guarantee LastModifiedDate strictly after the POST row. _, err := db.ExecContext(context.Background(), `UPDATE _System_ConnectorPushAttempt SET LastModifiedDate = DATE_ADD(NOW(), INTERVAL 1 SECOND) WHERE Id = 'attempt-t18-post'`) if err != nil { t.Fatalf("T18 backdate POST attempt: %v", err) } insertSyncAttemptWithMethod(t, db, "attempt-t18-patch", linkKey, ConnectorPushAttemptStatusTerminalFailed, "PATCH", 2, 500, "upstream returned 500", owner, slot, atype, obj, rec, dest, target) // Ensure PATCH LastModifiedDate > POST LastModifiedDate by updating PATCH to NOW()+2s. _, err = db.ExecContext(context.Background(), `UPDATE _System_ConnectorPushAttempt SET LastModifiedDate = DATE_ADD(NOW(), INTERVAL 2 SECOND) WHERE Id = 'attempt-t18-patch'`) if err != nil { t.Fatalf("T18 advance PATCH attempt: %v", err) } store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("T18: %v", err) } if len(results) != 1 { t.Fatalf("T18: expected 1 row; got %d: %+v", len(results), results) } r := results[0] if !r.HasLink { t.Errorf("T18: HasLink=false, want true") } if r.LinkStatus != ConnectorPushLinkStatusFinalized { t.Errorf("T18: LinkStatus=%q, want %q", r.LinkStatus, ConnectorPushLinkStatusFinalized) } if !r.HasAttempt { t.Errorf("T18: HasAttempt=false, want true (PATCH terminal_failed must be surfaced)") } if r.LastAttemptStatus != ConnectorPushAttemptStatusTerminalFailed { t.Errorf("T18: LastAttemptStatus=%q, want %q (PATCH failure must win latest-wins)", r.LastAttemptStatus, ConnectorPushAttemptStatusTerminalFailed) } if r.DestinationSlotID != dest { t.Errorf("T18: DestinationSlotID=%q, want %q", r.DestinationSlotID, dest) } } // T14: Empty result → empty slice (not nil), verified by checking len=0. func TestFindRecordPushState_T14_Empty(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{"Account", "no-such-record"}) if err != nil { t.Fatalf("T14: %v", err) } // Must be a non-nil empty slice (Go make([]T, 0, 0) is fine). if results == nil { t.Fatalf("T14: results is nil, want empty (non-nil) slice") } if len(results) != 0 { t.Fatalf("T14: expected 0 results; got %d: %+v", len(results), results) } } // TestFindRecordPushState_RetiredFinalizedHistoryDoesNotResurrect is the v16 // regression guard: one retired link + one finalized POST attempt for the same // 5-tuple, no active link → must return 0 rows. // // Pre-fix: the finalized POST attempt was in attemptMap and the retired link was // excluded by the SQL status filter (finalized+suspended_drift only), leaving the // attempt key in the union and producing a phantom attempt-only row. // Post-fix: finalized POST attempts are excluded from attemptLiveMap; without an // active link the 5-tuple emits no row. func TestFindRecordPushState_RetiredFinalizedHistoryDoesNotResurrect(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-v16" owner = "owner-v16" slot = "slot-v16" dest = "dest-v16" target = "contacts" linkKey = "lk-v16" atype = "PushToConnector" ) // Retired link: excluded by Query A status filter (finalized+suspended_drift only). insertSyncLink(t, db, "link-v16-retired", linkKey, ConnectorPushLinkStatusRetired, "remote-v16", owner, slot, atype, obj, rec, dest, target) // Finalized POST attempt: the original successful create. Excluded from // attemptLiveMap because Status='finalized'. insertSyncAttempt(t, db, "att-v16-finalized", linkKey, ConnectorPushAttemptStatusFinalized, 1, 200, "", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("v16: %v", err) } // No active link + only finalized history → no row emitted. if len(results) != 0 { t.Fatalf("v16: expected 0 rows (retired link + finalized attempt = no active destination); got %d: %+v", len(results), results) } } // TestFindRecordPushState_InFlightCreateSurfacesWithoutLink confirms that a // pending POST attempt (no link row) produces exactly one row with // HasLink=false, HasAttempt=true, LastAttemptStatus="pending". func TestFindRecordPushState_InFlightCreateSurfacesWithoutLink(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-inflight" owner = "owner-inflight" slot = "slot-inflight" dest = "dest-inflight" target = "contacts" linkKey = "lk-inflight" atype = "PushToConnector" ) // No link row. One pending POST attempt (in-flight create). insertSyncAttempt(t, db, "att-inflight", linkKey, ConnectorPushAttemptStatusPending, 1, 0, "", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("inflight: %v", err) } if len(results) != 1 { t.Fatalf("inflight: expected 1 row; got %d: %+v", len(results), results) } r := results[0] if r.HasLink { t.Errorf("inflight: HasLink=true, want false (no link row)") } if !r.HasAttempt { t.Errorf("inflight: HasAttempt=false, want true") } if r.LastAttemptStatus != ConnectorPushAttemptStatusPending { t.Errorf("inflight: LastAttemptStatus=%q, want %q", r.LastAttemptStatus, ConnectorPushAttemptStatusPending) } } // TestFindRecordPushState_OrphanPatchAttemptDoesNotSurface confirms that a // PATCH attempt with no link row does not produce any row, regardless of its // status. An orphan PATCH with no active link is stale history that must not // resurrect the destination lineage. func TestFindRecordPushState_OrphanPatchAttemptDoesNotSurface(t *testing.T) { db, cleanup := openSyncStatusTiDB(t) defer cleanup() const ( obj = "Lead" rec = "00Q-patch-orphan" owner = "owner-patch-orphan" slot = "slot-patch-orphan" dest = "dest-patch-orphan" target = "contacts" linkKey = "lk-patch-orphan" atype = "PushToConnector" ) // No link row. One terminal_failed PATCH attempt (orphan update attempt). insertSyncAttemptWithMethod(t, db, "att-patch-orphan", linkKey, ConnectorPushAttemptStatusTerminalFailed, "PATCH", 1, 500, "upstream gone", owner, slot, atype, obj, rec, dest, target) store := NewSQLConnectorPushDurableStore(db) results, err := store.FindRecordPushState(context.Background(), ConnectorPushRecordStateQuery{obj, rec}) if err != nil { t.Fatalf("patch-orphan: %v", err) } // Orphan PATCH: excluded from attemptLiveMap (not POST); no active link; no row. if len(results) != 0 { t.Fatalf("patch-orphan: expected 0 rows (orphan PATCH with no link must not surface); got %d: %+v", len(results), results) } } // TestFindRecordPushState_NoLeakColumns confirms that both SQL query constants // never reference any secret/body/credential/raw-error column. // The test scans findRecordPushStateQueryLink and findRecordPushStateQueryAttempt // directly so it genuinely fails if a prohibited column is added. func TestFindRecordPushState_NoLeakColumns(t *testing.T) { prohibited := []string{ // Secret / body / credential / raw-error columns. "LastError,", // bare LastError (not LastErrorAt) "RequestBodyBase64", "ResponseBodyBase64", "RequestHeaders", "ResponseHeaders", "CredentialRef", "ClaimToken", "OriginatingUserSessionJSON", "RetryPolicy", "RequestBodyHash", "ResponseBodyHash", "ProfileContractHash", "RemoteIdMirror", "RemoteIdProvenanceMirror", // Config-identity columns that must NOT appear in SELECT lists. // (ConnectorOwnerKey/ActionType ARE selected as group keys — that's by design. // ActionSlotId, LinkKey, RemoteId, ConnectorNamespace must NOT appear.) "ActionSlotId", "l.RemoteId", // link_remote_id "link_remote_id", "ConnectorNamespace", } for _, col := range prohibited { if strings.Contains(findRecordPushStateQueryLink, col) { t.Errorf("findRecordPushStateQueryLink must not reference %q", col) } if strings.Contains(findRecordPushStateQueryAttempt, col) { t.Errorf("findRecordPushStateQueryAttempt must not reference %q", col) } } }