Fix flaky CT submission bug

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## HEAD
 
+### Fix flaky submission failures
+* #1084: CT flaky submission failures
+
 ### Add support for WASI port
 * Add build tags for wasip1 GOOS
 

ctpolicy/chromepolicy.go

@@ -22,8 +22,8 @@ import (
 )
 
 const (
-	minOperators = 2                   // minimum number of distinct CT log operators that issue an SCT.
-	dayDuration  = 86400 * time.Second // time.Duration of one day
+	dayDuration          = 24 * time.Hour // time.Duration of one day
+	minDistinctOperators = 2              // Number of distinct CT log operators that submit an SCT
 )
 
 // ChromeCTPolicy implements logic for complying with Chrome's CT log policy
@@ -56,7 +56,7 @@ func (chromeP ChromeCTPolicy) LogsByGroup(cert *x509.Certificate, approved *logl
 	if err != nil {
 		return nil, err
 	}
-	baseGroup.MinOperators = minOperators
+	baseGroup.MinDistinctOperators = minDistinctOperators
 	groups[baseGroup.Name] = baseGroup
 	return groups, nil
 }

ctpolicy/chromepolicy_test.go

@@ -61,9 +61,9 @@ func wantedGroups(base int, minusBob bool) LogPolicyData {
 				"https://ct.googleapis.com/racketeer/":      true,
 				"https://log.bob.io":                        true,
 			},
-			MinInclusions: base,
-			MinOperators:  minOperators,
-			IsBase:        true,
+			MinInclusions:        base,
+			MinDistinctOperators: minDistinctOperators,
+			IsBase:               true,
 			LogWeights: map[string]float32{
 				"https://ct.googleapis.com/logs/argon2020/": 1.0,
 				"https://ct.googleapis.com/aviator/":        1.0,

ctpolicy/ctpolicy.go

@@ -31,13 +31,13 @@ const (
 
 // LogGroupInfo holds information on a single group of logs specified by Policy.
 type LogGroupInfo struct {
-	Name          string
-	LogURLs       map[string]bool    // set of members
-	MinInclusions int                // Required number of submissions.
-	MinOperators  int                // Required number of distinct CT log operators.
-	IsBase        bool               // True only for Log-group covering all logs.
-	LogWeights    map[string]float32 // weights used for submission, default weight is 1
-	wMu           sync.RWMutex       // guards weights
+	Name                 string
+	LogURLs              map[string]bool    // set of members
+	MinInclusions        int                // Required number of submissions.
+	MinDistinctOperators int                // Required number of distinct CT log operators that submit an SCT.
+	IsBase               bool               // True only for Log-group covering all logs.
+	LogWeights           map[string]float32 // weights used for submission, default weight is 1
+	wMu                  sync.RWMutex       // guards weights
 }
 
 func (group *LogGroupInfo) setMinInclusions(i int) error {

submission/races.go

@@ -52,27 +52,24 @@ type groupState struct {
 // When some group is complete cancels all requests that are not needed by any
 // group.
 type safeSubmissionState struct {
-	mu          sync.Mutex
-	logToGroups map[string]ctpolicy.GroupSet
-	groupNeeds  map[string]int // number of logs that need to be submitted for each group.
-	minGroups   int            // minimum number of distinct groups that need a log submitted.
+	mu                   sync.Mutex
+	logToGroups          map[string]ctpolicy.GroupSet
+	remainingSubmissions int             // number of logs that still need to be submitted.
+	minDistinctGroups    int             // number of groups that need a submission
+	groupsSubmitted      map[string]bool // number of logs submitted to each group.
 
-	groups  map[string]bool // set of groups that have a log submitted.
 	results map[string]*submissionResult
 	cancels map[string]context.CancelFunc
 }
 
 func newSafeSubmissionState(groups ctpolicy.LogPolicyData) *safeSubmissionState {
 	var s safeSubmissionState
 	s.logToGroups = ctpolicy.GroupByLogs(groups)
-	s.groupNeeds = make(map[string]int)
-	for _, g := range groups {
-		s.groupNeeds[g.Name] = g.MinInclusions
-	}
 	if baseGroup, ok := groups[ctpolicy.BaseName]; ok {
-		s.minGroups = baseGroup.MinOperators
+		s.remainingSubmissions = baseGroup.MinInclusions
+		s.minDistinctGroups = baseGroup.MinDistinctOperators
 	}
-	s.groups = make(map[string]bool)
+	s.groupsSubmitted = make(map[string]bool)
 	s.results = make(map[string]*submissionResult)
 	s.cancels = make(map[string]context.CancelFunc)
 	return &s
@@ -88,15 +85,7 @@ func (sub *safeSubmissionState) request(logURL string, cancel context.CancelFunc
 		return false
 	}
 	sub.results[logURL] = &submissionResult{}
-	isAwaited := false
-	for g := range sub.logToGroups[logURL] {
-		if sub.groupNeeds[g] > 0 {
-			isAwaited = true
-			break
-		}
-	}
-	if !isAwaited {
-		// No groups expecting result from this Log.
+	if sub.remainingSubmissions <= 0 {
 		return false
 	}
 	sub.cancels[logURL] = cancel
@@ -113,69 +102,59 @@ func (sub *safeSubmissionState) setResult(logURL string, sct *ct.SignedCertifica
 		sub.results[logURL] = &submissionResult{sct: sct, err: err}
 		return
 	}
+	// group name associated with logURL outside of BaseName.
+	// (this assumes the logURL is associated with only one group ignoring BaseName)
 	// If at least one group needs that SCT, result is set. Otherwise dumped.
 	for groupName := range sub.logToGroups[logURL] {
 		// Ignore the base group (All-logs) here to check separately.
 		if groupName == ctpolicy.BaseName {
 			continue
 		}
-		if sub.groupNeeds[groupName] > 0 {
+		// Set the result if the group does not have a submission.
+		if !sub.groupsSubmitted[groupName] {
 			sub.results[logURL] = &submissionResult{sct: sct, err: err}
+			sub.groupsSubmitted[groupName] = true
 		}
-		sub.groups[groupName] = true
-		sub.groupNeeds[groupName]--
 	}
 
 	// Check the base group (All-logs) only
 	if sub.logToGroups[logURL][ctpolicy.BaseName] {
 		if sub.results[logURL].sct != nil {
-			// It is already processed in a non-base group, so we can reduce the groupNeeds for the base group as well.
-			sub.groupNeeds[ctpolicy.BaseName]--
-		} else if sub.groupNeeds[ctpolicy.BaseName] > 0 {
-			minInclusionsForOtherGroup := 0
-			for g, cnt := range sub.groupNeeds {
-				if g != ctpolicy.BaseName && cnt > 0 {
-					minInclusionsForOtherGroup += cnt
-				}
-			}
+			// The cert has been observed in a non-base group, so account for it.
+			sub.remainingSubmissions--
+		} else if sub.remainingSubmissions > 0 {
+			// Arriving at this portion of the code implies that the result contains an SCT from
+			// the same log operator.
+			// reservedSubmissions represents the number of submissions that still need to be
+			// submitted from different log operators.
+			reservedSubmissions := sub.minDistinctGroups - len(sub.groupsSubmitted)
 			// Set the result only if the base group still needs SCTs more than total counts
 			// of minimum inclusions for other groups.
-			if sub.groupNeeds[ctpolicy.BaseName] > minInclusionsForOtherGroup {
+			if sub.remainingSubmissions > reservedSubmissions {
 				sub.results[logURL] = &submissionResult{sct: sct, err: err}
-				sub.groupNeeds[ctpolicy.BaseName]--
+				sub.remainingSubmissions--
 			}
 		}
 	}
 
 	// Cancel any pending Log-requests for which there're no more awaiting
 	// Log-groups.
-	for logURL, groupSet := range sub.logToGroups {
-		isAwaited := false
-		for g := range groupSet {
-			if sub.groupNeeds[g] > 0 {
-				isAwaited = true
-				break
-			}
-		}
-		if !isAwaited && sub.cancels[logURL] != nil {
+	for logURL := range sub.logToGroups {
+		if sub.remainingSubmissions <= 0 && sub.cancels[logURL] != nil {
 			sub.cancels[logURL]()
 			sub.cancels[logURL] = nil
 		}
 	}
 }
 
 // groupComplete returns true iff the specified group has all the SCTs it needs.
-func (sub *safeSubmissionState) groupComplete(groupName string) bool {
+func (sub *safeSubmissionState) groupComplete() bool {
 	sub.mu.Lock()
 	defer sub.mu.Unlock()
-	needs, ok := sub.groupNeeds[groupName]
-	if !ok {
-		return true
-	}
-	if len(sub.groups) < sub.minGroups {
+	if len(sub.groupsSubmitted) < sub.minDistinctGroups {
 		return false
 	}
-	return needs <= 0
+	return sub.remainingSubmissions <= 0
 }
 
 func (sub *safeSubmissionState) collectSCTs() []*AssignedSCT {
@@ -226,7 +205,7 @@ func groupRace(ctx context.Context, chain []ct.ASN1Cert, asPreChain bool,
 				return
 			case <-timeoutchan:
 			}
-			if state.groupComplete(group.Name) {
+			if state.groupComplete() {
 				cancel()
 				return
 			}
@@ -243,14 +222,14 @@ func groupRace(ctx context.Context, chain []ct.ASN1Cert, asPreChain bool,
 	for range session {
 		select {
 		case <-ctx.Done():
-			return groupState{Name: group.Name, Success: state.groupComplete(group.Name)}
+			return groupState{Name: group.Name, Success: state.groupComplete()}
 		case <-counter:
-			if state.groupComplete(group.Name) {
+			if state.groupComplete() {
 				return groupState{Name: group.Name, Success: true}
 			}
 		}
 	}
-	return groupState{Name: group.Name, Success: state.groupComplete(group.Name)}
+	return groupState{Name: group.Name, Success: state.groupComplete()}
 }
 
 func parallelNums(groups ctpolicy.LogPolicyData) map[string]int {

submission/races_test.go

@@ -93,7 +93,7 @@ func TestGetSCTs(t *testing.T) {
 			name:   "singleGroupOneSCT",
 			sbMock: &mockSubmitter{fixedDelay: map[byte]time.Duration{'a': 0}, firstLetterURLReqNumber: make(map[byte]int)},
 			groups: ctpolicy.LogPolicyData{
-				"a": {
+				ctpolicy.BaseName: {
 					Name:          "a",
 					LogURLs:       map[string]bool{"a1.com": true, "a2.com": true},
 					MinInclusions: 1,
@@ -107,7 +107,7 @@ func TestGetSCTs(t *testing.T) {
 			name:   "singleGroupMultiSCT",
 			sbMock: &mockSubmitter{fixedDelay: map[byte]time.Duration{'a': 0}, firstLetterURLReqNumber: make(map[byte]int)},
 			groups: ctpolicy.LogPolicyData{
-				"a": {
+				ctpolicy.BaseName: {
 					Name:          "a",
 					LogURLs:       map[string]bool{"a1.com": true, "a2.com": true, "a3.com": true, "a4.com": true, "a5.com": true},
 					MinInclusions: 3,
@@ -124,27 +124,27 @@ func TestGetSCTs(t *testing.T) {
 				"a": {
 					Name:          "a",
 					LogURLs:       map[string]bool{"a1.com": true, "a2.com": true, "a3.com": true, "a4.com": true},
-					MinInclusions: 1,
+					MinInclusions: 0,
 					IsBase:        false,
 					LogWeights:    map[string]float32{"a1.com": 1.0, "a2.com": 1.0, "a3.com": 1.0, "a4.com": 1.0},
 				},
 				"b": {
 					Name:          "b",
 					LogURLs:       map[string]bool{"b1.com": true, "b2.com": true, "b3.com": true, "b4.com": true},
-					MinInclusions: 1,
+					MinInclusions: 0,
 					IsBase:        false,
 					LogWeights:    map[string]float32{"b1.com": 1.0, "b2.com": 1.0, "b3.com": 1.0, "b4.com": 1.0},
 				},
 				ctpolicy.BaseName: {
-					Name:          ctpolicy.BaseName,
-					LogURLs:       map[string]bool{"a1.com": true, "a2.com": true, "a3.com": true, "a4.com": true, "b1.com": true, "b2.com": true, "b3.com": true, "b4.com": true},
-					MinInclusions: 5,
-					MinOperators:  2,
-					IsBase:        true,
-					LogWeights:    map[string]float32{"a1.com": 1.0, "a2.com": 1.0, "a3.com": 1.0, "a4.com": 1.0, "b1.com": 1.0, "b2.com": 1.0, "b3.com": 1.0, "b4.com": 1.0},
+					Name:                 ctpolicy.BaseName,
+					LogURLs:              map[string]bool{"a1.com": true, "a2.com": true, "a3.com": true, "a4.com": true, "b1.com": true, "b2.com": true, "b3.com": true, "b4.com": true},
+					MinInclusions:        2,
+					MinDistinctOperators: 2,
+					IsBase:               true,
+					LogWeights:           map[string]float32{"a1.com": 1.0, "a2.com": 1.0, "a3.com": 1.0, "a4.com": 1.0, "b1.com": 1.0, "b2.com": 1.0, "b3.com": 1.0, "b4.com": 1.0},
 				},
 			},
-			resultTrail: map[string]int{"a": 1, "b": 1, ctpolicy.BaseName: 5},
+			resultTrail: map[string]int{"a": 1, "b": 1, ctpolicy.BaseName: 2},
 		},
 		{
 			name:   "notEnoughDistinctOperators",
@@ -158,12 +158,12 @@ func TestGetSCTs(t *testing.T) {
 					LogWeights:    map[string]float32{"a1.com": 1.0, "a2.com": 1.0, "a3.com": 1.0, "a4.com": 1.0},
 				},
 				ctpolicy.BaseName: {
-					Name:          ctpolicy.BaseName,
-					LogURLs:       map[string]bool{"a1.com": true, "a2.com": true, "a3.com": true, "a4.com": true},
-					MinInclusions: 2,
-					MinOperators:  2,
-					IsBase:        true,
-					LogWeights:    map[string]float32{"a1.com": 1.0, "a2.com": 1.0, "a3.com": 1.0, "a4.com": 1.0},
+					Name:                 ctpolicy.BaseName,
+					LogURLs:              map[string]bool{"a1.com": true, "a2.com": true, "a3.com": true, "a4.com": true},
+					MinInclusions:        2,
+					MinDistinctOperators: 2,
+					IsBase:               true,
+					LogWeights:           map[string]float32{"a1.com": 1.0, "a2.com": 1.0, "a3.com": 1.0, "a4.com": 1.0},
 				},
 			},
 			errRegexp: regexp.MustCompile("didn't receive enough SCTs"),