55 "encoding/json"
66 "fmt"
77 "math"
8+ "net/http"
9+ "slices"
10+ "strconv"
811 "sync"
912 "testing"
1013 "time"
@@ -23,6 +26,7 @@ import (
2326
2427 backendconfig "github.com/rudderlabs/rudder-server/backend-config"
2528 "github.com/rudderlabs/rudder-server/enterprise/reporting"
29+ "github.com/rudderlabs/rudder-server/jobsdb"
2630 mocksRouter "github.com/rudderlabs/rudder-server/mocks/router"
2731 mocksTransformer "github.com/rudderlabs/rudder-server/mocks/router/transformer"
2832 "github.com/rudderlabs/rudder-server/processor/integrations"
@@ -35,6 +39,7 @@ import (
3539 transformerFeaturesService "github.com/rudderlabs/rudder-server/services/transformer"
3640 "github.com/rudderlabs/rudder-server/services/transientsource"
3741 "github.com/rudderlabs/rudder-server/utils/cache"
42+ utilTypes "github.com/rudderlabs/rudder-server/utils/types"
3843)
3944
4045// createTestWorker creates a worker instance for testing with properly initialized StatsCache instances
@@ -55,6 +60,212 @@ func createTestWorker(destType string, transformProxy bool, stat stats.Stats) *w
5560 }
5661}
5762
63+ // TestDeliveredWithWarningsEnabled covers the OR between the destination-definition capability
64+ // (the GA switch) and the per-workspace controlled-rollout allow-list.
65+ func TestDeliveredWithWarningsEnabled (t * testing.T ) {
66+ const (
67+ enabledWorkspace = "workspace-enabled"
68+ otherWorkspace = "workspace-other"
69+ )
70+
71+ newHandle := func (destDefSupports bool , allowList ... string ) * Handle {
72+ rt := & Handle {}
73+ rt .supportsDeliveredWithWarnings .Store (destDefSupports )
74+ rt .deliveredWithWarningsEnabledForWorkspace = func (workspaceID string ) bool {
75+ return slices .Contains (allowList , workspaceID )
76+ }
77+ return rt
78+ }
79+
80+ t .Run ("destination definition supports it, workspace not allow-listed" , func (t * testing.T ) {
81+ require .True (t , newHandle (true ).deliveredWithWarningsEnabled (otherWorkspace ))
82+ })
83+
84+ t .Run ("destination definition does not support it, workspace allow-listed" , func (t * testing.T ) {
85+ require .True (t , newHandle (false , enabledWorkspace ).deliveredWithWarningsEnabled (enabledWorkspace ))
86+ })
87+
88+ t .Run ("both enabled" , func (t * testing.T ) {
89+ require .True (t , newHandle (true , enabledWorkspace ).deliveredWithWarningsEnabled (enabledWorkspace ))
90+ })
91+
92+ t .Run ("neither enabled" , func (t * testing.T ) {
93+ require .False (t , newHandle (false , enabledWorkspace ).deliveredWithWarningsEnabled (otherWorkspace ))
94+ })
95+ }
96+
97+ // TestGateDeliveredWithWarning covers the 296 -> 200 downgrade applied per job, per workspace.
98+ func TestGateDeliveredWithWarning (t * testing.T ) {
99+ const (
100+ gateDestType = "BRAZE"
101+ downgradeMetric = "router_status_downgraded_count"
102+ enabledWorkspace = "workspace-enabled"
103+ otherWorkspace = "workspace-other"
104+ )
105+ downgradeTags := stats.Tags {
106+ "destType" : gateDestType ,
107+ "from" : strconv .Itoa (utilTypes .DeliveredWithWarningCode ),
108+ "to" : strconv .Itoa (utilTypes .SuccessEventCode ),
109+ }
110+
111+ newGateWorker := func (t * testing.T , destDefSupports bool , allowList ... string ) (* worker , * memstats.Store ) {
112+ t .Helper ()
113+ statsStore , err := memstats .New ()
114+ require .NoError (t , err )
115+ rt := & Handle {
116+ destType : gateDestType ,
117+ statusDowngradedStat : func (from , to int ) stats.Counter {
118+ return statsStore .NewTaggedStat (downgradeMetric , stats .CountType , stats.Tags {
119+ "destType" : gateDestType ,
120+ "from" : strconv .Itoa (from ),
121+ "to" : strconv .Itoa (to ),
122+ })
123+ },
124+ }
125+ rt .supportsDeliveredWithWarnings .Store (destDefSupports )
126+ rt .deliveredWithWarningsEnabledForWorkspace = func (workspaceID string ) bool {
127+ return slices .Contains (allowList , workspaceID )
128+ }
129+ return & worker {rt : rt , logger : logger .NOP }, statsStore
130+ }
131+
132+ downgrades := func (statsStore * memstats.Store ) float64 {
133+ if m := statsStore .Get (downgradeMetric , downgradeTags ); m != nil {
134+ return m .LastValue ()
135+ }
136+ return 0
137+ }
138+
139+ jobsOf := func (metadata ... types.JobMetadataT ) types.DestinationJobT {
140+ return types.DestinationJobT {JobMetadataArray : metadata }
141+ }
142+ jobMeta := func (jobID int64 , workspaceID string ) types.JobMetadataT {
143+ return types.JobMetadataT {JobID : jobID , WorkspaceID : workspaceID }
144+ }
145+
146+ t .Run ("allow-listed workspace keeps 296" , func (t * testing.T ) {
147+ w , statsStore := newGateWorker (t , false , enabledWorkspace )
148+ codes := map [int64 ]int {1 : utilTypes .DeliveredWithWarningCode }
149+ w .gateDeliveredWithWarning (jobsOf (jobMeta (1 , enabledWorkspace )), codes )
150+ require .Equal (t , utilTypes .DeliveredWithWarningCode , codes [1 ])
151+ require .Zero (t , downgrades (statsStore ))
152+ })
153+
154+ t .Run ("destination definition support keeps 296 for any workspace" , func (t * testing.T ) {
155+ w , statsStore := newGateWorker (t , true )
156+ codes := map [int64 ]int {1 : utilTypes .DeliveredWithWarningCode }
157+ w .gateDeliveredWithWarning (jobsOf (jobMeta (1 , otherWorkspace )), codes )
158+ require .Equal (t , utilTypes .DeliveredWithWarningCode , codes [1 ])
159+ require .Zero (t , downgrades (statsStore ))
160+ })
161+
162+ t .Run ("non allow-listed workspace downgrades to 200" , func (t * testing.T ) {
163+ w , statsStore := newGateWorker (t , false , enabledWorkspace )
164+ codes := map [int64 ]int {1 : utilTypes .DeliveredWithWarningCode }
165+ w .gateDeliveredWithWarning (jobsOf (jobMeta (1 , otherWorkspace )), codes )
166+ require .Equal (t , http .StatusOK , codes [1 ])
167+ require .EqualValues (t , 1 , downgrades (statsStore ))
168+ })
169+
170+ t .Run ("batch spanning workspaces downgrades only the non allow-listed job" , func (t * testing.T ) {
171+ w , statsStore := newGateWorker (t , false , enabledWorkspace )
172+ codes := map [int64 ]int {
173+ 1 : utilTypes .DeliveredWithWarningCode ,
174+ 2 : utilTypes .DeliveredWithWarningCode ,
175+ }
176+ w .gateDeliveredWithWarning (jobsOf (jobMeta (1 , enabledWorkspace ), jobMeta (2 , otherWorkspace )), codes )
177+ require .Equal (t , utilTypes .DeliveredWithWarningCode , codes [1 ])
178+ require .Equal (t , http .StatusOK , codes [2 ])
179+ require .EqualValues (t , 1 , downgrades (statsStore ))
180+ })
181+
182+ t .Run ("only 296 is rewritten in a mixed-code batch" , func (t * testing.T ) {
183+ w , statsStore := newGateWorker (t , false )
184+ codes := map [int64 ]int {
185+ 1 : http .StatusOK ,
186+ 2 : utilTypes .DeliveredWithWarningCode ,
187+ 3 : http .StatusBadRequest ,
188+ 4 : http .StatusInternalServerError ,
189+ }
190+ w .gateDeliveredWithWarning (jobsOf (
191+ jobMeta (1 , otherWorkspace ), jobMeta (2 , otherWorkspace ),
192+ jobMeta (3 , otherWorkspace ), jobMeta (4 , otherWorkspace ),
193+ ), codes )
194+ require .Equal (t , map [int64 ]int {
195+ 1 : http .StatusOK ,
196+ 2 : http .StatusOK ,
197+ 3 : http .StatusBadRequest ,
198+ 4 : http .StatusInternalServerError ,
199+ }, codes )
200+ require .EqualValues (t , 1 , downgrades (statsStore ))
201+ })
202+
203+ t .Run ("duplicate job metadata downgrades once" , func (t * testing.T ) {
204+ w , statsStore := newGateWorker (t , false )
205+ codes := map [int64 ]int {1 : utilTypes .DeliveredWithWarningCode }
206+ w .gateDeliveredWithWarning (jobsOf (jobMeta (1 , otherWorkspace ), jobMeta (1 , otherWorkspace )), codes )
207+ require .Equal (t , http .StatusOK , codes [1 ])
208+ require .EqualValues (t , 1 , downgrades (statsStore ))
209+ })
210+ }
211+
212+ // TestPostStatusOnResponseQStoreDeliveredWithWarningPayload verifies that, on the success path,
213+ // only a Delivered-with-Warning (296) status reports the transformed delivery body instead of the
214+ // router input payload, gated by the storeDeliveredWithWarningPayload flag.
215+ func TestPostStatusOnResponseQStoreDeliveredWithWarningPayload (t * testing.T ) {
216+ const (
217+ destType = "BRAZE"
218+ inputPayload = `{"input":"event"}`
219+ deliveryBody = `{"delivery":"transformed batch body"}`
220+ )
221+
222+ newTestWorker := func (storePayload bool ) * worker {
223+ return & worker {
224+ logger : logger .NOP ,
225+ rt : & Handle {
226+ destType : destType ,
227+ responseQ : make (chan workerJobStatus , 1 ),
228+ reportJobsdbPayload : config .SingleValueLoader (true ),
229+ storeDeliveredWithWarningPayload : config .SingleValueLoader (storePayload ),
230+ },
231+ }
232+ }
233+
234+ report := func (w * worker , statusCode int , message string ) workerJobStatus {
235+ destinationJob := & types.DestinationJobT {Message : json .RawMessage (message )}
236+ metadata := & types.JobMetadataT {JobT : & jobsdb.JobT {EventPayload : json .RawMessage (inputPayload )}}
237+ status := & jobsdb.JobStatusT {ErrorResponse : json .RawMessage (`{}` )}
238+ w .postStatusOnResponseQ (statusCode , destinationJob , "application/json" , metadata , status , "" )
239+ return <- w .rt .responseQ
240+ }
241+
242+ t .Run ("296 reports the transformed delivery body and marks payloadStage=delivery" , func (t * testing.T ) {
243+ got := report (newTestWorker (true ), utilTypes .DeliveredWithWarningCode , deliveryBody )
244+ require .JSONEq (t , deliveryBody , string (got .payload ))
245+ require .Equal (t , jobsdb .Succeeded .State , got .status .JobState )
246+ require .Contains (t , string (got .status .ErrorResponse ), `"payloadStage":"delivery"` )
247+ })
248+
249+ t .Run ("200 keeps the router input payload with no delivery marker" , func (t * testing.T ) {
250+ got := report (newTestWorker (true ), utilTypes .SuccessEventCode , deliveryBody )
251+ require .JSONEq (t , inputPayload , string (got .payload ))
252+ require .Equal (t , jobsdb .Succeeded .State , got .status .JobState )
253+ require .NotContains (t , string (got .status .ErrorResponse ), "payloadStage" )
254+ })
255+
256+ t .Run ("296 with storeDeliveredWithWarningPayload off falls back to the input payload" , func (t * testing.T ) {
257+ got := report (newTestWorker (false ), utilTypes .DeliveredWithWarningCode , deliveryBody )
258+ require .JSONEq (t , inputPayload , string (got .payload ))
259+ require .NotContains (t , string (got .status .ErrorResponse ), "payloadStage" )
260+ })
261+
262+ t .Run ("filter event code stays Filtered with the input payload" , func (t * testing.T ) {
263+ got := report (newTestWorker (true ), utilTypes .FilterEventCode , deliveryBody )
264+ require .JSONEq (t , inputPayload , string (got .payload ))
265+ require .Equal (t , jobsdb .Filtered .State , got .status .JobState )
266+ })
267+ }
268+
58269func TestConsolidateRespBodys (t * testing.T ) {
59270 tcs := []struct {
60271 in []map [int64 ]string
0 commit comments