automod: action limits; create reports for interaction churn (#465)

The motivation here is to start auto-reporting in production based on
specific rules. Specifically, this PR would start reporting on
interactions churn (follow/unfollow), letting human mods confirm before
taking action on an account.

Before we do that, need to de-duplicate reports. For example, if an
account creates thousands of spammy posts, only want to report the
account once. Generally want to prevent run-away rules from creating
millions of reports, or doing thousands of automated account takedowns
(for example).

This PR prevents re-reporting based on daily counters (fast checks), as
well as double-checking against the mod service API just before filing a
report (slower, but reliable).

This PR also adds "quotas" for mod actions, implemented using the
counter system. This isn't perfect (the counter system itself might be
buggy or broken (causing duplicate actions), but seems like a good
start. If quotas are exceeded, automod will log and skip taking
additional actions until the next day.

automod/action_dedupe_test.go

@@ -0,0 +1,45 @@
+package automod
+
+import (
+	"context"
+	"testing"
+
+	appbsky "github.com/bluesky-social/indigo/api/bsky"
+	"github.com/bluesky-social/indigo/atproto/identity"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func alwaysReportAccountRule(evt *RecordEvent) error {
+	evt.ReportAccount(ReportReasonOther, "test report")
+	return nil
+}
+
+func TestAccountReportDedupe(t *testing.T) {
+	assert := assert.New(t)
+	ctx := context.Background()
+	engine := engineFixture()
+	engine.Rules = RuleSet{
+		RecordRules: []RecordRuleFunc{
+			alwaysReportAccountRule,
+		},
+	}
+
+	path := "app.bsky.feed.post/abc123"
+	cid1 := "cid123"
+	p1 := appbsky.FeedPost{Text: "some post blah"}
+	id1 := identity.Identity{
+		DID:    syntax.DID("did:plc:abc111"),
+		Handle: syntax.Handle("handle.example.com"),
+	}
+
+	// exact same event multiple times; should only report once
+	for i := 0; i < 5; i++ {
+		assert.NoError(engine.ProcessRecord(ctx, id1.DID, path, cid1, &p1))
+	}
+
+	reports, err := engine.GetCount("automod-quota", "report", PeriodDay)
+	assert.NoError(err)
+	assert.Equal(1, reports)
+}

automod/circuit_breaker_test.go

@@ -0,0 +1,94 @@
+package automod
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	appbsky "github.com/bluesky-social/indigo/api/bsky"
+	"github.com/bluesky-social/indigo/atproto/identity"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func alwaysTakedownRecordRule(evt *RecordEvent) error {
+	evt.TakedownRecord()
+	return nil
+}
+
+func alwaysReportRecordRule(evt *RecordEvent) error {
+	evt.ReportRecord(ReportReasonOther, "test report")
+	return nil
+}
+
+func TestTakedownCircuitBreaker(t *testing.T) {
+	assert := assert.New(t)
+	ctx := context.Background()
+	engine := engineFixture()
+	dir := identity.NewMockDirectory()
+	engine.Directory = &dir
+	// note that this is a record-level action, not account-level
+	engine.Rules = RuleSet{
+		RecordRules: []RecordRuleFunc{
+			alwaysTakedownRecordRule,
+		},
+	}
+
+	path := "app.bsky.feed.post/abc123"
+	cid1 := "cid123"
+	p1 := appbsky.FeedPost{Text: "some post blah"}
+
+	// generate double the quote of events; expect to only count the quote worth of actions
+	for i := 0; i < 2*QuotaModTakedownDay; i++ {
+		ident := identity.Identity{
+			DID:    syntax.DID(fmt.Sprintf("did:plc:abc%d", i)),
+			Handle: syntax.Handle("handle.example.com"),
+		}
+		dir.Insert(ident)
+		assert.NoError(engine.ProcessRecord(ctx, ident.DID, path, cid1, &p1))
+	}
+
+	takedowns, err := engine.GetCount("automod-quota", "takedown", PeriodDay)
+	assert.NoError(err)
+	assert.Equal(QuotaModTakedownDay, takedowns)
+
+	reports, err := engine.GetCount("automod-quota", "report", PeriodDay)
+	assert.NoError(err)
+	assert.Equal(0, reports)
+}
+
+func TestReportCircuitBreaker(t *testing.T) {
+	assert := assert.New(t)
+	ctx := context.Background()
+	engine := engineFixture()
+	dir := identity.NewMockDirectory()
+	engine.Directory = &dir
+	engine.Rules = RuleSet{
+		RecordRules: []RecordRuleFunc{
+			alwaysReportRecordRule,
+		},
+	}
+
+	path := "app.bsky.feed.post/abc123"
+	cid1 := "cid123"
+	p1 := appbsky.FeedPost{Text: "some post blah"}
+
+	// generate double the quota of events; expect to only count the quota worth of actions
+	for i := 0; i < 2*QuotaModReportDay; i++ {
+		ident := identity.Identity{
+			DID:    syntax.DID(fmt.Sprintf("did:plc:abc%d", i)),
+			Handle: syntax.Handle("handle.example.com"),
+		}
+		dir.Insert(ident)
+		assert.NoError(engine.ProcessRecord(ctx, ident.DID, path, cid1, &p1))
+	}
+
+	takedowns, err := engine.GetCount("automod-quota", "takedown", PeriodDay)
+	assert.NoError(err)
+	assert.Equal(0, takedowns)
+
+	reports, err := engine.GetCount("automod-quota", "report", PeriodDay)
+	assert.NoError(err)
+	assert.Equal(QuotaModReportDay, reports)
+}

automod/countstore/countstore_test.go

@@ -20,9 +20,12 @@ func TestMemCountStoreBasics(t *testing.T) {
 	assert.Equal(0, c)
 	assert.NoError(cs.Increment(ctx, "test1", "val1"))
 	assert.NoError(cs.Increment(ctx, "test1", "val1"))
-	c, err = cs.GetCount(ctx, "test1", "val1", PeriodTotal)
-	assert.NoError(err)
-	assert.Equal(2, c)
+
+	for _, period := range []string{PeriodTotal, PeriodDay, PeriodHour} {
+		c, err = cs.GetCount(ctx, "test1", "val1", period)
+		assert.NoError(err)
+		assert.Equal(2, c)
+	}
 
 	c, err = cs.GetCountDistinct(ctx, "test2", "val2", PeriodTotal)
 	assert.NoError(err)
@@ -36,9 +39,12 @@ func TestMemCountStoreBasics(t *testing.T) {
 
 	assert.NoError(cs.IncrementDistinct(ctx, "test2", "val2", "two"))
 	assert.NoError(cs.IncrementDistinct(ctx, "test2", "val2", "three"))
-	c, err = cs.GetCountDistinct(ctx, "test2", "val2", PeriodTotal)
-	assert.NoError(err)
-	assert.Equal(3, c)
+
+	for _, period := range []string{PeriodTotal, PeriodDay, PeriodHour} {
+		c, err = cs.GetCountDistinct(ctx, "test2", "val2", period)
+		assert.NoError(err)
+		assert.Equal(3, c)
+	}
 }
 
 func TestMemCountStoreConcurrent(t *testing.T) {

automod/engine.go

@@ -72,6 +72,10 @@ func (e *Engine) ProcessIdentityEvent(ctx context.Context, t string, did syntax.
 	if err := evt.PersistCounters(ctx); err != nil {
 		return err
 	}
+	// check for any new errors during persist
+	if evt.Err != nil {
+		return evt.Err
+	}
 	return nil
 }
 
@@ -114,6 +118,10 @@ func (e *Engine) ProcessRecord(ctx context.Context, did syntax.DID, path, recCID
 	if err := evt.PersistCounters(ctx); err != nil {
 		return err
 	}
+	// check for any new errors during persist
+	if evt.Err != nil {
+		return evt.Err
+	}
 	return nil
 }