this repo has no description
make automod event processing timeouts configurable (#918)
+103
-61
+25
-13
automod/engine/engine.go
+25
-13
automod/engine/engine.go
···
17
17
"github.com/bluesky-social/indigo/xrpc"
18
18
)
19
19
20
-
const (
21
-
recordEventTimeout = 30 * time.Second
22
-
identityEventTimeout = 10 * time.Second
23
-
notificationEventTimeout = 5 * time.Second
24
-
)
25
-
26
20
// runtime for executing rules, managing state, and recording moderation actions.
27
21
//
28
22
// NOTE: careful when initializing: several fields must not be nil or zero, even though they are pointer type.
···
60
54
QuotaModTakedownDay int
61
55
// number of misc actions automod can do per day, for all subjects combined (circuit breaker)
62
56
QuotaModActionDay int
57
+
58
+
// timeout for record event processing (total, including all setup, rules, and teardown)
59
+
RecordEventTimeout time.Duration
60
+
// timeout for identity event and account event processing (total, including all setup, rules, and teardown)
61
+
IdentityEventTimeout time.Duration
62
+
// timeout for event processing (total, including all setup, rules, and teardown)
63
+
OzoneEventTimeout time.Duration
63
64
}
64
65
65
66
// Entrypoint for external code pushing #identity events in to the engine.
···
85
86
eventErrorCount.WithLabelValues("identity").Inc()
86
87
}
87
88
}()
88
-
ctx, cancel := context.WithTimeout(ctx, identityEventTimeout)
89
-
defer cancel()
89
+
var cancel context.CancelFunc
90
+
if eng.Config.IdentityEventTimeout != 0 {
91
+
ctx, cancel = context.WithTimeout(ctx, eng.Config.IdentityEventTimeout)
92
+
defer cancel()
93
+
}
90
94
91
95
// first purge any caches; we need to re-resolve from scratch on identity updates
92
96
if err := eng.PurgeAccountCaches(ctx, did); err != nil {
···
156
160
eventErrorCount.WithLabelValues("account").Inc()
157
161
}
158
162
}()
159
-
ctx, cancel := context.WithTimeout(ctx, identityEventTimeout)
160
-
defer cancel()
163
+
var cancel context.CancelFunc
164
+
if eng.Config.IdentityEventTimeout != 0 {
165
+
ctx, cancel = context.WithTimeout(ctx, eng.Config.IdentityEventTimeout)
166
+
defer cancel()
167
+
}
161
168
162
169
// first purge any caches; we need to re-resolve from scratch on account updates
163
170
if err := eng.PurgeAccountCaches(ctx, did); err != nil {
···
221
228
eng.Logger.Error("automod event execution exception", "err", r, "did", op.DID, "collection", op.Collection, "rkey", op.RecordKey)
222
229
}
223
230
}()
224
-
ctx, cancel := context.WithTimeout(ctx, recordEventTimeout)
225
-
defer cancel()
231
+
var cancel context.CancelFunc
232
+
if eng.Config.RecordEventTimeout != 0 {
233
+
ctx, cancel = context.WithTimeout(ctx, eng.Config.RecordEventTimeout)
234
+
defer cancel()
235
+
}
226
236
227
237
if err := op.Validate(); err != nil {
228
238
eventErrorCount.WithLabelValues("record").Inc()
···
287
297
}
288
298
289
299
// returns a boolean indicating "block the event"
300
+
// NOTE: this code is unused and should be removed
290
301
func (eng *Engine) ProcessNotificationEvent(ctx context.Context, senderDID, recipientDID syntax.DID, reason string, subject syntax.ATURI) (bool, error) {
291
302
eventProcessCount.WithLabelValues("notif").Inc()
292
303
start := time.Now()
···
301
312
eng.Logger.Error("automod event execution exception", "err", r, "sender", senderDID, "recipient", recipientDID)
302
313
}
303
314
}()
304
-
ctx, cancel := context.WithTimeout(ctx, notificationEventTimeout)
315
+
var cancel context.CancelFunc
316
+
ctx, cancel = context.WithTimeout(ctx, time.Second*5)
305
317
defer cancel()
306
318
307
319
senderIdent, err := eng.Directory.LookupDID(ctx, senderDID)
+5
-2
automod/engine/engine_ozone.go
+5
-2
automod/engine/engine_ozone.go
···
163
163
eng.Logger.Error("automod ozone event execution exception", "err", r, "eventID", eventView.Id, "createdAt", eventView.CreatedAt)
164
164
}
165
165
}()
166
-
ctx, cancel := context.WithTimeout(ctx, recordEventTimeout)
167
-
defer cancel()
166
+
var cancel context.CancelFunc
167
+
if eng.Config.OzoneEventTimeout != 0 {
168
+
ctx, cancel = context.WithTimeout(ctx, eng.Config.OzoneEventTimeout)
169
+
defer cancel()
170
+
}
168
171
169
172
ec, err := NewOzoneEventContext(ctx, eng, eventView)
170
173
if err != nil {
+42
-21
cmd/hepa/main.go
+42
-21
cmd/hepa/main.go
···
173
173
EnvVars: []string{"HEPA_QUOTA_MOD_ACTION_DAY"},
174
174
Value: 2000,
175
175
},
176
+
&cli.DurationFlag{
177
+
Name: "record-event-timeout",
178
+
Usage: "total processing time for record events (including setup, rules, and persisting)",
179
+
EnvVars: []string{"HEPA_RECORD_EVENT_TIMEOUT"},
180
+
Value: 30 * time.Second,
181
+
},
182
+
&cli.DurationFlag{
183
+
Name: "identity-event-timeout",
184
+
Usage: "total processing time for identity and account events (including setup, rules, and persisting)",
185
+
EnvVars: []string{"HEPA_IDENTITY_EVENT_TIMEOUT"},
186
+
Value: 10 * time.Second,
187
+
},
188
+
&cli.DurationFlag{
189
+
Name: "ozone-event-timeout",
190
+
Usage: "total processing time for ozone events (including setup, rules, and persisting)",
191
+
EnvVars: []string{"HEPA_OZONE_EVENT_TIMEOUT"},
192
+
Value: 30 * time.Second,
193
+
},
176
194
}
177
195
178
196
app.Commands = []*cli.Command{
···
260
278
srv, err := NewServer(
261
279
dir,
262
280
Config{
263
-
Logger: logger,
264
-
BskyHost: cctx.String("atp-bsky-host"),
265
-
OzoneHost: cctx.String("atp-ozone-host"),
266
-
OzoneDID: cctx.String("ozone-did"),
267
-
OzoneAdminToken: cctx.String("ozone-admin-token"),
268
-
PDSHost: cctx.String("atp-pds-host"),
269
-
PDSAdminToken: cctx.String("pds-admin-token"),
270
-
SetsFileJSON: cctx.String("sets-json-path"),
271
-
RedisURL: cctx.String("redis-url"),
272
-
SlackWebhookURL: cctx.String("slack-webhook-url"),
273
-
HiveAPIToken: cctx.String("hiveai-api-token"),
274
-
AbyssHost: cctx.String("abyss-host"),
275
-
AbyssPassword: cctx.String("abyss-password"),
276
-
RatelimitBypass: cctx.String("ratelimit-bypass"),
277
-
RulesetName: cctx.String("ruleset"),
278
-
PreScreenHost: cctx.String("prescreen-host"),
279
-
PreScreenToken: cctx.String("prescreen-token"),
280
-
ReportDupePeriod: cctx.Duration("report-dupe-period"),
281
-
QuotaModReportDay: cctx.Int("quota-mod-report-day"),
282
-
QuotaModTakedownDay: cctx.Int("quota-mod-takedown-day"),
283
-
QuotaModActionDay: cctx.Int("quota-mod-action-day"),
281
+
Logger: logger,
282
+
BskyHost: cctx.String("atp-bsky-host"),
283
+
OzoneHost: cctx.String("atp-ozone-host"),
284
+
OzoneDID: cctx.String("ozone-did"),
285
+
OzoneAdminToken: cctx.String("ozone-admin-token"),
286
+
PDSHost: cctx.String("atp-pds-host"),
287
+
PDSAdminToken: cctx.String("pds-admin-token"),
288
+
SetsFileJSON: cctx.String("sets-json-path"),
289
+
RedisURL: cctx.String("redis-url"),
290
+
SlackWebhookURL: cctx.String("slack-webhook-url"),
291
+
HiveAPIToken: cctx.String("hiveai-api-token"),
292
+
AbyssHost: cctx.String("abyss-host"),
293
+
AbyssPassword: cctx.String("abyss-password"),
294
+
RatelimitBypass: cctx.String("ratelimit-bypass"),
295
+
RulesetName: cctx.String("ruleset"),
296
+
PreScreenHost: cctx.String("prescreen-host"),
297
+
PreScreenToken: cctx.String("prescreen-token"),
298
+
ReportDupePeriod: cctx.Duration("report-dupe-period"),
299
+
QuotaModReportDay: cctx.Int("quota-mod-report-day"),
300
+
QuotaModTakedownDay: cctx.Int("quota-mod-takedown-day"),
301
+
QuotaModActionDay: cctx.Int("quota-mod-action-day"),
302
+
RecordEventTimeout: cctx.Duration("record-event-timeout"),
303
+
IdentityEventTimeout: cctx.Duration("identity-event-timeout"),
304
+
OzoneEventTimeout: cctx.Duration("ozone-event-timeout"),
284
305
},
285
306
)
286
307
if err != nil {
+31
-25
cmd/hepa/server.go
+31
-25
cmd/hepa/server.go
···
33
33
}
34
34
35
35
type Config struct {
36
-
Logger *slog.Logger
37
-
BskyHost string
38
-
OzoneHost string
39
-
OzoneDID string
40
-
OzoneAdminToken string
41
-
PDSHost string
42
-
PDSAdminToken string
43
-
SetsFileJSON string
44
-
RedisURL string
45
-
SlackWebhookURL string
46
-
HiveAPIToken string
47
-
AbyssHost string
48
-
AbyssPassword string
49
-
RulesetName string
50
-
RatelimitBypass string
51
-
PreScreenHost string
52
-
PreScreenToken string
53
-
ReportDupePeriod time.Duration
54
-
QuotaModReportDay int
55
-
QuotaModTakedownDay int
56
-
QuotaModActionDay int
36
+
Logger *slog.Logger
37
+
BskyHost string
38
+
OzoneHost string
39
+
OzoneDID string
40
+
OzoneAdminToken string
41
+
PDSHost string
42
+
PDSAdminToken string
43
+
SetsFileJSON string
44
+
RedisURL string
45
+
SlackWebhookURL string
46
+
HiveAPIToken string
47
+
AbyssHost string
48
+
AbyssPassword string
49
+
RulesetName string
50
+
RatelimitBypass string
51
+
PreScreenHost string
52
+
PreScreenToken string
53
+
ReportDupePeriod time.Duration
54
+
QuotaModReportDay int
55
+
QuotaModTakedownDay int
56
+
QuotaModActionDay int
57
+
RecordEventTimeout time.Duration
58
+
IdentityEventTimeout time.Duration
59
+
OzoneEventTimeout time.Duration
57
60
}
58
61
59
62
func NewServer(dir identity.Directory, config Config) (*Server, error) {
···
215
218
AdminClient: adminClient,
216
219
BlobClient: blobClient,
217
220
Config: engine.EngineConfig{
218
-
ReportDupePeriod: config.ReportDupePeriod,
219
-
QuotaModReportDay: config.QuotaModReportDay,
220
-
QuotaModTakedownDay: config.QuotaModTakedownDay,
221
-
QuotaModActionDay: config.QuotaModActionDay,
221
+
ReportDupePeriod: config.ReportDupePeriod,
222
+
QuotaModReportDay: config.QuotaModReportDay,
223
+
QuotaModTakedownDay: config.QuotaModTakedownDay,
224
+
QuotaModActionDay: config.QuotaModActionDay,
225
+
RecordEventTimeout: config.RecordEventTimeout,
226
+
IdentityEventTimeout: config.IdentityEventTimeout,
227
+
OzoneEventTimeout: config.OzoneEventTimeout,
222
228
},
223
229
}
224
230