diff --git a/api/schedule/v1/message.pb.go b/api/schedule/v1/message.pb.go
index f6cb3f25af6..c823ac27f11 100644
--- a/api/schedule/v1/message.pb.go
+++ b/api/schedule/v1/message.pb.go
@@ -572,9 +572,10 @@ func (*WatchWorkflowResponse) XXX_OneofWrappers() []interface{} {
 }
 
 type StartWorkflowRequest struct {
-	Request              *v14.StartWorkflowExecutionRequest `protobuf:"bytes,2,opt,name=request,proto3" json:"request,omitempty"`
-	LastCompletionResult *v11.Payloads                      `protobuf:"bytes,4,opt,name=last_completion_result,json=lastCompletionResult,proto3" json:"last_completion_result,omitempty"`
-	ContinuedFailure     *v12.Failure                       `protobuf:"bytes,5,opt,name=continued_failure,json=continuedFailure,proto3" json:"continued_failure,omitempty"`
+	Request                 *v14.StartWorkflowExecutionRequest `protobuf:"bytes,2,opt,name=request,proto3" json:"request,omitempty"`
+	LastCompletionResult    *v11.Payloads                      `protobuf:"bytes,4,opt,name=last_completion_result,json=lastCompletionResult,proto3" json:"last_completion_result,omitempty"`
+	ContinuedFailure        *v12.Failure                       `protobuf:"bytes,5,opt,name=continued_failure,json=continuedFailure,proto3" json:"continued_failure,omitempty"`
+	CompletedRateLimitSleep bool                               `protobuf:"varint,6,opt,name=completed_rate_limit_sleep,json=completedRateLimitSleep,proto3" json:"completed_rate_limit_sleep,omitempty"`
 }
 
 func (m *StartWorkflowRequest) Reset()      { *m = StartWorkflowRequest{} }
@@ -630,6 +631,13 @@ func (m *StartWorkflowRequest) GetContinuedFailure() *v12.Failure {
 	return nil
 }
 
+func (m *StartWorkflowRequest) GetCompletedRateLimitSleep() bool {
+	if m != nil {
+		return m.CompletedRateLimitSleep
+	}
+	return false
+}
+
 type StartWorkflowResponse struct {
 	RunId         string     `protobuf:"bytes,1,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"`
 	RealStartTime *time.Time `protobuf:"bytes,2,opt,name=real_start_time,json=realStartTime,proto3,stdtime" json:"real_start_time,omitempty"`
@@ -836,77 +844,79 @@ func init() {
 }
 
 var fileDescriptor_6461b6986ba20ee7 = []byte{
-	// 1117 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x56, 0x41, 0x6f, 0x1b, 0x45,
-	0x14, 0xf6, 0xda, 0x8e, 0x63, 0x8f, 0x63, 0x37, 0xd9, 0x26, 0x91, 0x15, 0x60, 0xe3, 0x5a, 0xb4,
-	0x04, 0x09, 0xd6, 0x4a, 0x2a, 0x71, 0x28, 0xa8, 0x52, 0x93, 0x12, 0x62, 0x04, 0xc2, 0x9a, 0x04,
-	0x82, 0xb8, 0xac, 0x26, 0xbb, 0x63, 0x67, 0xd5, 0xf1, 0xcc, 0x32, 0x33, 0x9b, 0x36, 0x9c, 0x38,
-	0x70, 0x45, 0xea, 0x1f, 0xe0, 0xc4, 0x85, 0x2b, 0x42, 0xf0, 0x1b, 0x38, 0xe6, 0xd8, 0x1b, 0xc4,
-	0xb9, 0x20, 0x71, 0xe9, 0x4f, 0x40, 0x33, 0x3b, 0xe3, 0xc4, 0x4e, 0xd2, 0x04, 0x8a, 0x84, 0xb8,
-	0xed, 0x7b, 0xfb, 0xbe, 0x6f, 0xde, 0xbe, 0xf7, 0xbd, 0x37, 0x0b, 0xde, 0x96, 0x78, 0x90, 0x30,
-	0x8e, 0x48, 0x5b, 0x60, 0x7e, 0x80, 0x79, 0x1b, 0x25, 0x71, 0x5b, 0x84, 0xfb, 0x38, 0x4a, 0x09,
-	0x6e, 0x1f, 0xac, 0xb6, 0x07, 0x58, 0x08, 0xd4, 0xc7, 0x7e, 0xc2, 0x99, 0x64, 0xee, 0xb2, 0x0d,
-	0xf7, 0xb3, 0x70, 0x1f, 0x25, 0xb1, 0x6f, 0xc3, 0xfd, 0x83, 0xd5, 0xa5, 0xd7, 0x47, 0x7c, 0x8a,
-	0x28, 0x64, 0x83, 0x01, 0xa3, 0xe7, 0x68, 0x96, 0x6e, 0x8f, 0x45, 0xf5, 0x50, 0x4c, 0x52, 0x7e,
-	0xfe, 0xb4, 0x09, 0x32, 0x4c, 0xd3, 0x81, 0x50, 0x41, 0xa3, 0xf3, 0x5e, 0x18, 0xf5, 0x98, 0xf1,
-	0x47, 0x3d, 0xc2, 0x1e, 0x9b, 0xa8, 0x3b, 0x63, 0x51, 0x97, 0x7e, 0xe1, 0xd2, 0x3b, 0x63, 0x71,
-	0x96, 0x44, 0x7d, 0x6d, 0x1c, 0xea, 0x70, 0x8e, 0xbf, 0x4c, 0xb1, 0x90, 0x01, 0xc7, 0x22, 0x61,
-	0x54, 0x58, 0xdc, 0x72, 0x9f, 0xb1, 0x3e, 0xc1, 0x6d, 0x6d, 0xed, 0xa5, 0xbd, 0xb6, 0x8c, 0x07,
-	0x58, 0x48, 0x34, 0x48, 0x4c, 0xc0, 0xad, 0x08, 0x27, 0x98, 0x46, 0x98, 0x86, 0x31, 0x16, 0xed,
-	0x3e, 0xeb, 0x33, 0xed, 0xd7, 0x4f, 0x59, 0x48, 0xeb, 0x9b, 0x3c, 0xa8, 0xad, 0xa7, 0xbd, 0x1e,
-	0xe6, 0x38, 0xda, 0x96, 0x88, 0x4b, 0x77, 0x03, 0xcc, 0x50, 0x36, 0x88, 0x29, 0x22, 0x81, 0xe2,
-	0x6b, 0x38, 0x4d, 0x67, 0xa5, 0xba, 0xb6, 0xe4, 0x67, 0x87, 0xf9, 0xf6, 0x30, 0x7f, 0xc7, 0x1e,
-	0xb6, 0x5e, 0x7c, 0xfa, 0xdb, 0xb2, 0x03, 0xab, 0x06, 0xa5, 0xfc, 0xee, 0x03, 0x50, 0x45, 0xa1,
-	0x4c, 0x2d, 0x47, 0xfe, 0x9a, 0x1c, 0x20, 0x03, 0x69, 0x8a, 0x6d, 0x50, 0x67, 0x07, 0x98, 0x13,
-	0x94, 0x04, 0x09, 0x23, 0x71, 0x78, 0xd8, 0x28, 0x34, 0x9d, 0x95, 0xfa, 0xda, 0x5b, 0xfe, 0x48,
-	0x10, 0x4a, 0x09, 0xba, 0xf8, 0xfe, 0xc1, 0xaa, 0xbf, 0x6d, 0xea, 0xfb, 0x49, 0x06, 0xea, 0x6a,
-	0x0c, 0xac, 0xb1, 0xb3, 0xa6, 0xbb, 0x08, 0x4a, 0x03, 0x44, 0x53, 0x44, 0x1a, 0xc5, 0xa6, 0xb3,
-	0x52, 0x86, 0xc6, 0x6a, 0x7d, 0x5b, 0x04, 0xb5, 0x0e, 0x95, 0x98, 0x53, 0x44, 0xb6, 0x25, 0x92,
-	0xd8, 0x7d, 0x15, 0x54, 0x28, 0x1a, 0x60, 0x91, 0xa0, 0x30, 0xab, 0x41, 0x05, 0x9e, 0x3a, 0xdc,
-	0x5b, 0x60, 0x66, 0x64, 0x04, 0x71, 0xa4, 0x3f, 0xb0, 0x02, 0xab, 0x23, 0x5f, 0x27, 0x72, 0x97,
-	0x41, 0xd5, 0xb6, 0x5c, 0x45, 0x94, 0x75, 0x04, 0xb0, 0xae, 0x4e, 0xe4, 0x76, 0xc1, 0x4d, 0x82,
-	0x84, 0x0c, 0x12, 0xce, 0x42, 0x2c, 0x04, 0x8e, 0xb2, 0x5a, 0x15, 0xae, 0x59, 0xab, 0x39, 0x05,
-	0xee, 0x5a, 0xac, 0x2e, 0xd9, 0x2e, 0xb8, 0xb1, 0x67, 0x7a, 0x19, 0x08, 0xd5, 0x4c, 0xd1, 0x28,
-	0x36, 0x0b, 0x2b, 0xd5, 0x35, 0xdf, 0xbf, 0x62, 0x88, 0xfc, 0x31, 0x0d, 0xc0, 0xfa, 0xde, 0x59,
-	0x53, 0xb8, 0x9f, 0x81, 0x45, 0x9d, 0x6a, 0xc8, 0x06, 0x09, 0xc1, 0x32, 0x66, 0x54, 0x69, 0x31,
-	0x25, 0xb2, 0x31, 0xa5, 0xb3, 0x6d, 0x8e, 0xf7, 0x24, 0x9b, 0x41, 0x45, 0xdb, 0x45, 0x87, 0x84,
-	0xa1, 0x48, 0xc0, 0x79, 0x85, 0xdf, 0x18, 0xc1, 0xa1, 0x46, 0xbb, 0x1f, 0x83, 0xb9, 0x90, 0x51,
-	0x19, 0xd3, 0x14, 0x47, 0x81, 0x99, 0xc9, 0x46, 0xe9, 0x22, 0x4a, 0xf3, 0x52, 0x71, 0x6e, 0x66,
-	0x8f, 0x70, 0x76, 0x04, 0x35, 0x1e, 0xf7, 0x36, 0xa8, 0x87, 0x8c, 0xf6, 0x48, 0x1c, 0xca, 0x40,
-	0xb2, 0x47, 0x98, 0x36, 0xa6, 0x9b, 0xce, 0x4a, 0x01, 0xd6, 0xac, 0x77, 0x47, 0x39, 0x75, 0xf3,
-	0x30, 0x8e, 0x02, 0x8e, 0x7b, 0x1c, 0x8b, 0xfd, 0x46, 0x45, 0x4b, 0xa1, 0xaa, 0x7c, 0x30, 0x73,
-	0xb5, 0xbe, 0xcf, 0x83, 0x39, 0xfd, 0xed, 0x56, 0x55, 0x0f, 0x78, 0x5f, 0xb8, 0xf7, 0x41, 0xd9,
-	0xd6, 0xcc, 0x8c, 0x45, 0x6b, 0x3c, 0xcb, 0xb3, 0x15, 0xb5, 0x48, 0x38, 0xc2, 0xb8, 0xf7, 0x40,
-	0x31, 0xa6, 0x3d, 0x66, 0xc6, 0xe1, 0xce, 0xd5, 0xd8, 0x0e, 0xed, 0x31, 0xa8, 0x31, 0xee, 0x47,
-	0xa0, 0x16, 0xd3, 0x58, 0xc6, 0x88, 0x04, 0x09, 0x92, 0xe1, 0xbe, 0xd1, 0xc9, 0x1b, 0x57, 0x93,
-	0x74, 0x55, 0x38, 0x9c, 0x31, 0x68, 0x6d, 0xb9, 0x0f, 0xc1, 0x94, 0x50, 0x32, 0xd7, 0x63, 0x70,
-	0x1d, 0x7d, 0x8c, 0x0d, 0x07, 0xcc, 0xc0, 0xad, 0xaf, 0xc0, 0xdc, 0x66, 0x4a, 0xc8, 0xa7, 0x49,
-	0xa4, 0x9c, 0xd9, 0x92, 0x7a, 0xe9, 0x22, 0x9d, 0x6f, 0x62, 0xfe, 0x82, 0x26, 0xb6, 0x7e, 0x76,
-	0xc0, 0xec, 0x43, 0x2c, 0x42, 0x1e, 0xef, 0x61, 0x68, 0xf6, 0xe2, 0x7f, 0xda, 0xa0, 0xf3, 0x79,
-	0x17, 0x2e, 0xca, 0xfb, 0x47, 0x07, 0xcc, 0xef, 0xaa, 0x1e, 0xec, 0x9a, 0x3d, 0x6f, 0xeb, 0xf6,
-	0x01, 0xa8, 0xe0, 0x27, 0x38, 0x4c, 0xd5, 0x78, 0x98, 0xe6, 0xbe, 0x79, 0xd9, 0x58, 0x59, 0xec,
-	0xfb, 0x16, 0x00, 0x4f, 0xb1, 0xee, 0x5d, 0xb0, 0xd8, 0x8b, 0xb9, 0x90, 0xc1, 0xc8, 0x15, 0xf0,
-	0x94, 0xaa, 0x1d, 0x54, 0xd4, 0x3b, 0xe8, 0xa6, 0x7e, 0x7b, 0x0a, 0x4d, 0x69, 0x27, 0x72, 0x5f,
-	0x01, 0x15, 0xc2, 0x68, 0x5f, 0xad, 0x5a, 0xa2, 0x87, 0xba, 0x0c, 0xcb, 0xca, 0xd1, 0x65, 0x84,
-	0xb4, 0xfe, 0x74, 0xc0, 0xc2, 0x44, 0xce, 0xa6, 0xe0, 0x9b, 0xa0, 0xa4, 0xa4, 0x90, 0x0a, 0x5d,
-	0xee, 0xfa, 0x59, 0x21, 0x8d, 0x2d, 0xe7, 0x73, 0x09, 0x6f, 0x6b, 0x14, 0x34, 0x68, 0xf7, 0x1e,
-	0x28, 0x99, 0x85, 0x92, 0xbf, 0xde, 0x42, 0xd9, 0xca, 0x41, 0x83, 0x70, 0xdf, 0x03, 0xd3, 0x76,
-	0x75, 0x14, 0xae, 0xb7, 0x3a, 0xb6, 0x72, 0xd0, 0x42, 0xd6, 0x67, 0x41, 0x3d, 0xe3, 0xb1, 0xfb,
-	0xa7, 0xf5, 0x5d, 0x1e, 0xcc, 0xeb, 0xd9, 0x9f, 0xec, 0xd0, 0xe7, 0x60, 0xda, 0xdc, 0xc4, 0x26,
-	0xcb, 0xfb, 0xe3, 0x07, 0x4d, 0xdc, 0xdc, 0x5a, 0x27, 0x67, 0x79, 0x4e, 0x4b, 0x9e, 0xb1, 0x40,
-	0x4b, 0xf7, 0x82, 0xfd, 0x5a, 0xfc, 0xf7, 0xf7, 0xeb, 0xd4, 0x3f, 0xdd, 0xaf, 0x1f, 0x16, 0xcb,
-	0x85, 0xd9, 0x62, 0xeb, 0x09, 0x58, 0x98, 0x28, 0x8f, 0x11, 0xc3, 0x02, 0x28, 0x19, 0xa1, 0x65,
-	0xf7, 0xe5, 0x14, 0xd7, 0xd2, 0xda, 0x02, 0x37, 0x38, 0x46, 0x24, 0xbb, 0x91, 0xfe, 0xde, 0xff,
-	0x40, 0x4d, 0x01, 0xf5, 0x61, 0xea, 0x4d, 0xeb, 0x27, 0x07, 0x2c, 0x6c, 0x20, 0x1a, 0x62, 0x32,
-	0xd9, 0x9a, 0xd7, 0x00, 0xb0, 0x3f, 0x49, 0x71, 0xa4, 0x65, 0x50, 0x81, 0x15, 0xe3, 0xe9, 0x44,
-	0xee, 0x12, 0x28, 0xc7, 0x11, 0xa6, 0x32, 0x96, 0x87, 0x66, 0x08, 0x46, 0xf6, 0xf8, 0xdc, 0x4d,
-	0xbd, 0xc4, 0xdc, 0x2d, 0x2a, 0x0d, 0x23, 0xc1, 0xa8, 0xbe, 0xc1, 0x2a, 0xd0, 0x58, 0xad, 0x5f,
-	0x1c, 0xd0, 0xd8, 0xc1, 0x5c, 0xfd, 0x1c, 0x49, 0xfc, 0x3f, 0x4a, 0x7c, 0x3d, 0x3a, 0x3a, 0xf6,
-	0x72, 0xcf, 0x8e, 0xbd, 0xdc, 0xf3, 0x63, 0xcf, 0xf9, 0x7a, 0xe8, 0x39, 0x3f, 0x0c, 0x3d, 0xe7,
-	0xd7, 0xa1, 0xe7, 0x1c, 0x0d, 0x3d, 0xe7, 0xf7, 0xa1, 0xe7, 0xfc, 0x31, 0xf4, 0x72, 0xcf, 0x87,
-	0x9e, 0xf3, 0xf4, 0xc4, 0xcb, 0x1d, 0x9d, 0x78, 0xb9, 0x67, 0x27, 0x5e, 0xee, 0x0b, 0xbf, 0xcf,
-	0x4e, 0xb3, 0x88, 0xd9, 0x25, 0x3f, 0xf9, 0xef, 0xda, 0xe7, 0xbd, 0x92, 0xee, 0xfe, 0xdd, 0xbf,
-	0x02, 0x00, 0x00, 0xff, 0xff, 0xa8, 0x9f, 0xa4, 0x54, 0x17, 0x0c, 0x00, 0x00,
+	// 1150 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x57, 0x4f, 0x6f, 0x1b, 0xc5,
+	0x1b, 0xf6, 0xda, 0x8e, 0x6b, 0x8f, 0x63, 0x37, 0xd9, 0x26, 0xf9, 0x59, 0xfe, 0xc1, 0xc6, 0xb5,
+	0x68, 0x09, 0x12, 0xac, 0x95, 0x54, 0xe2, 0xd0, 0xa2, 0x4a, 0x4d, 0x4a, 0x88, 0x51, 0x11, 0xd6,
+	0x38, 0x10, 0xc4, 0x65, 0x35, 0xd9, 0x1d, 0x3b, 0xa3, 0xce, 0xce, 0x2c, 0x3b, 0xb3, 0x69, 0xc3,
+	0x89, 0x03, 0x57, 0xa4, 0x7e, 0x07, 0x2e, 0x5c, 0x11, 0x82, 0xcf, 0xc0, 0x31, 0xc7, 0x8a, 0x0b,
+	0xc4, 0xb9, 0x20, 0x71, 0xe9, 0x47, 0x40, 0x33, 0x3b, 0xeb, 0xc4, 0x4e, 0xd2, 0x04, 0x8a, 0x84,
+	0xb8, 0xf9, 0x7d, 0xe7, 0x7d, 0x9e, 0x79, 0xf7, 0xfd, 0xf3, 0xec, 0x1a, 0xbc, 0x23, 0x71, 0x18,
+	0xf1, 0x18, 0xd1, 0x8e, 0xc0, 0xf1, 0x3e, 0x8e, 0x3b, 0x28, 0x22, 0x1d, 0xe1, 0xef, 0xe1, 0x20,
+	0xa1, 0xb8, 0xb3, 0xbf, 0xda, 0x09, 0xb1, 0x10, 0x68, 0x88, 0xdd, 0x28, 0xe6, 0x92, 0xdb, 0xcb,
+	0x59, 0xb8, 0x9b, 0x86, 0xbb, 0x28, 0x22, 0x6e, 0x16, 0xee, 0xee, 0xaf, 0x36, 0xdf, 0x18, 0xf3,
+	0x29, 0x22, 0x9f, 0x87, 0x21, 0x67, 0x67, 0x68, 0x9a, 0xb7, 0x26, 0xa2, 0x06, 0x88, 0xd0, 0x24,
+	0x3e, 0x7b, 0xdb, 0x14, 0x19, 0x66, 0x49, 0x28, 0x54, 0xd0, 0xf8, 0xbe, 0x97, 0x46, 0x3d, 0xe1,
+	0xf1, 0xe3, 0x01, 0xe5, 0x4f, 0x4c, 0xd4, 0xed, 0x89, 0xa8, 0x0b, 0x9f, 0xb0, 0xf9, 0xee, 0x44,
+	0x5c, 0x46, 0xa2, 0x9e, 0x96, 0xf8, 0x3a, 0x3c, 0xc6, 0x5f, 0x24, 0x58, 0x48, 0x2f, 0xc6, 0x22,
+	0xe2, 0x4c, 0x64, 0xb8, 0xe5, 0x21, 0xe7, 0x43, 0x8a, 0x3b, 0xda, 0xda, 0x4d, 0x06, 0x1d, 0x49,
+	0x42, 0x2c, 0x24, 0x0a, 0x23, 0x13, 0x70, 0x33, 0xc0, 0x11, 0x66, 0x01, 0x66, 0x3e, 0xc1, 0xa2,
+	0x33, 0xe4, 0x43, 0xae, 0xfd, 0xfa, 0x57, 0x1a, 0xd2, 0xfe, 0x3a, 0x0f, 0x6a, 0xeb, 0xc9, 0x60,
+	0x80, 0x63, 0x1c, 0xf4, 0x25, 0x8a, 0xa5, 0xbd, 0x01, 0x66, 0x19, 0x0f, 0x09, 0x43, 0xd4, 0x53,
+	0x7c, 0x0d, 0xab, 0x65, 0xad, 0x54, 0xd7, 0x9a, 0x6e, 0x7a, 0x99, 0x9b, 0x5d, 0xe6, 0x6e, 0x67,
+	0x97, 0xad, 0x17, 0x9f, 0xfd, 0xba, 0x6c, 0xc1, 0xaa, 0x41, 0x29, 0xbf, 0xfd, 0x00, 0x54, 0x91,
+	0x2f, 0x93, 0x8c, 0x23, 0x7f, 0x45, 0x0e, 0x90, 0x82, 0x34, 0x45, 0x1f, 0xd4, 0xf9, 0x3e, 0x8e,
+	0x29, 0x8a, 0xbc, 0x88, 0x53, 0xe2, 0x1f, 0x34, 0x0a, 0x2d, 0x6b, 0xa5, 0xbe, 0xf6, 0xb6, 0x3b,
+	0x1e, 0x08, 0x35, 0x09, 0xba, 0xf8, 0xee, 0xfe, 0xaa, 0xdb, 0x37, 0xf5, 0xfd, 0x38, 0x05, 0xf5,
+	0x34, 0x06, 0xd6, 0xf8, 0x69, 0xd3, 0x5e, 0x02, 0xa5, 0x10, 0xb1, 0x04, 0xd1, 0x46, 0xb1, 0x65,
+	0xad, 0x94, 0xa1, 0xb1, 0xda, 0xdf, 0x14, 0x41, 0xad, 0xcb, 0x24, 0x8e, 0x19, 0xa2, 0x7d, 0x89,
+	0x24, 0xb6, 0x5f, 0x03, 0x15, 0x86, 0x42, 0x2c, 0x22, 0xe4, 0xa7, 0x35, 0xa8, 0xc0, 0x13, 0x87,
+	0x7d, 0x13, 0xcc, 0x8e, 0x0d, 0x8f, 0x04, 0xfa, 0x01, 0x2b, 0xb0, 0x3a, 0xf6, 0x75, 0x03, 0x7b,
+	0x19, 0x54, 0xb3, 0x96, 0xab, 0x88, 0xb2, 0x8e, 0x00, 0x99, 0xab, 0x1b, 0xd8, 0x3d, 0x70, 0x83,
+	0x22, 0x21, 0xbd, 0x28, 0xe6, 0x3e, 0x16, 0x02, 0x07, 0x69, 0xad, 0x0a, 0x57, 0xac, 0xd5, 0xbc,
+	0x02, 0xf7, 0x32, 0xac, 0x2e, 0xd9, 0x0e, 0xb8, 0xbe, 0x6b, 0x7a, 0xe9, 0x09, 0xd5, 0x4c, 0xd1,
+	0x28, 0xb6, 0x0a, 0x2b, 0xd5, 0x35, 0xd7, 0xbd, 0x64, 0x89, 0xdc, 0x89, 0x19, 0x80, 0xf5, 0xdd,
+	0xd3, 0xa6, 0xb0, 0x3f, 0x05, 0x4b, 0x3a, 0x55, 0x9f, 0x87, 0x11, 0xc5, 0x92, 0x70, 0xa6, 0x66,
+	0x31, 0xa1, 0xb2, 0x31, 0xa3, 0xb3, 0x6d, 0x4d, 0xf6, 0x24, 0xdd, 0x41, 0x45, 0xdb, 0x43, 0x07,
+	0x94, 0xa3, 0x40, 0xc0, 0x05, 0x85, 0xdf, 0x18, 0xc3, 0xa1, 0x46, 0xdb, 0x1f, 0x81, 0x79, 0x9f,
+	0x33, 0x49, 0x58, 0x82, 0x03, 0xcf, 0xec, 0x64, 0xa3, 0x74, 0x1e, 0xa5, 0x39, 0x54, 0x9c, 0x9b,
+	0xe9, 0x4f, 0x38, 0x37, 0x86, 0x1a, 0x8f, 0x7d, 0x0b, 0xd4, 0x7d, 0xce, 0x06, 0x94, 0xf8, 0xd2,
+	0x93, 0xfc, 0x31, 0x66, 0x8d, 0x6b, 0x2d, 0x6b, 0xa5, 0x00, 0x6b, 0x99, 0x77, 0x5b, 0x39, 0x75,
+	0xf3, 0x30, 0x0e, 0xbc, 0x18, 0x0f, 0x62, 0x2c, 0xf6, 0x1a, 0x15, 0x3d, 0x0a, 0x55, 0xe5, 0x83,
+	0xa9, 0xab, 0xfd, 0x6d, 0x1e, 0xcc, 0xeb, 0x67, 0xcf, 0xa6, 0xea, 0x41, 0x3c, 0x14, 0xf6, 0x7d,
+	0x50, 0xce, 0x6a, 0x66, 0xd6, 0xa2, 0x3d, 0x99, 0xe5, 0xe9, 0x8a, 0x66, 0x48, 0x38, 0xc6, 0xd8,
+	0x77, 0x41, 0x91, 0xb0, 0x01, 0x37, 0xeb, 0x70, 0xfb, 0x72, 0x6c, 0x97, 0x0d, 0x38, 0xd4, 0x18,
+	0xfb, 0x11, 0xa8, 0x11, 0x46, 0x24, 0x41, 0xd4, 0x8b, 0x90, 0xf4, 0xf7, 0xcc, 0x9c, 0xbc, 0x79,
+	0x39, 0x49, 0x4f, 0x85, 0xc3, 0x59, 0x83, 0xd6, 0x96, 0xfd, 0x10, 0xcc, 0x08, 0x35, 0xe6, 0x7a,
+	0x0d, 0xae, 0x32, 0x1f, 0x13, 0xcb, 0x01, 0x53, 0x70, 0xfb, 0x4b, 0x30, 0xbf, 0x99, 0x50, 0xfa,
+	0x49, 0x14, 0x28, 0x67, 0x2a, 0x52, 0xaf, 0x5c, 0xa4, 0xb3, 0x4d, 0xcc, 0x9f, 0xd3, 0xc4, 0xf6,
+	0x8f, 0x16, 0x98, 0x7b, 0x88, 0x85, 0x1f, 0x93, 0x5d, 0x0c, 0x8d, 0x2e, 0xfe, 0xab, 0x0d, 0x3a,
+	0x9b, 0x77, 0xe1, 0xbc, 0xbc, 0xbf, 0xb7, 0xc0, 0xc2, 0x8e, 0xea, 0xc1, 0x8e, 0xd1, 0xf9, 0xac,
+	0x6e, 0x1f, 0x80, 0x0a, 0x7e, 0x8a, 0xfd, 0x44, 0xad, 0x87, 0x69, 0xee, 0x5b, 0x17, 0xad, 0x55,
+	0x86, 0x7d, 0x3f, 0x03, 0xc0, 0x13, 0xac, 0x7d, 0x07, 0x2c, 0x0d, 0x48, 0x2c, 0xa4, 0x37, 0x76,
+	0x79, 0x71, 0xc2, 0x94, 0x06, 0x15, 0xb5, 0x06, 0xdd, 0xd0, 0xa7, 0x27, 0xd0, 0x84, 0x75, 0x03,
+	0xfb, 0xff, 0xa0, 0x42, 0x39, 0x1b, 0x2a, 0xa9, 0xa5, 0x7a, 0xa9, 0xcb, 0xb0, 0xac, 0x1c, 0x3d,
+	0x4e, 0x69, 0xfb, 0x0f, 0x0b, 0x2c, 0x4e, 0xe5, 0x6c, 0x0a, 0xbe, 0x09, 0x4a, 0x6a, 0x14, 0x12,
+	0xa1, 0xcb, 0x5d, 0x3f, 0x3d, 0x48, 0x13, 0xe2, 0x7c, 0x26, 0xe1, 0xbe, 0x46, 0x41, 0x83, 0xb6,
+	0xef, 0x82, 0x92, 0x11, 0x94, 0xfc, 0xd5, 0x04, 0x65, 0x2b, 0x07, 0x0d, 0xc2, 0x7e, 0x0f, 0x5c,
+	0xcb, 0xa4, 0xa3, 0x70, 0x35, 0xe9, 0xd8, 0xca, 0xc1, 0x0c, 0xb2, 0x3e, 0x07, 0xea, 0x29, 0x4f,
+	0xa6, 0x3f, 0xed, 0x5f, 0xf2, 0x60, 0x41, 0xef, 0xfe, 0x74, 0x87, 0x3e, 0x03, 0xd7, 0xcc, 0x9b,
+	0xd8, 0x64, 0x79, 0x7f, 0xf2, 0xa2, 0xa9, 0x37, 0xb7, 0x9e, 0x93, 0xd3, 0x3c, 0x27, 0x25, 0x4f,
+	0x59, 0x60, 0x46, 0xf7, 0x12, 0x7d, 0x2d, 0xfe, 0xf3, 0xfa, 0x3a, 0xf3, 0xb7, 0xf5, 0xf5, 0x1e,
+	0x68, 0x9a, 0x0c, 0x95, 0x7a, 0x22, 0x89, 0x3d, 0x4a, 0x42, 0x22, 0x3d, 0x41, 0x31, 0x8e, 0xb4,
+	0x6e, 0x97, 0xe1, 0xff, 0xc6, 0x11, 0x10, 0x49, 0xfc, 0x48, 0x9d, 0xf7, 0xd5, 0xf1, 0x87, 0xc5,
+	0x72, 0x61, 0xae, 0xd8, 0x7e, 0x0a, 0x16, 0xa7, 0x6a, 0x6b, 0x26, 0x69, 0x11, 0x94, 0xcc, 0x94,
+	0xa6, 0x2f, 0xdb, 0x99, 0x58, 0xcf, 0xe5, 0x16, 0xb8, 0x1e, 0x63, 0x44, 0xd3, 0xd7, 0xd9, 0x5f,
+	0xfb, 0x98, 0xa8, 0x29, 0xa0, 0xbe, 0x4c, 0x9d, 0xb4, 0x7f, 0xb0, 0xc0, 0xe2, 0x06, 0x62, 0x3e,
+	0xa6, 0xd3, 0x7d, 0x7d, 0x1d, 0x80, 0xec, 0x0b, 0x8b, 0x04, 0x7a, 0x86, 0x2a, 0xb0, 0x62, 0x3c,
+	0xdd, 0xc0, 0x6e, 0x82, 0x32, 0x09, 0x30, 0x93, 0x44, 0x1e, 0x98, 0x0d, 0x1a, 0xdb, 0x93, 0x4b,
+	0x3b, 0xf3, 0x0a, 0x4b, 0xbb, 0xa4, 0x16, 0x00, 0x09, 0xce, 0x74, 0x19, 0x2b, 0xd0, 0x58, 0xed,
+	0x9f, 0x2c, 0xd0, 0xd8, 0xc6, 0xb1, 0xfa, 0xb2, 0x92, 0xf8, 0x3f, 0x94, 0xf8, 0x7a, 0x70, 0x78,
+	0xe4, 0xe4, 0x9e, 0x1f, 0x39, 0xb9, 0x17, 0x47, 0x8e, 0xf5, 0xd5, 0xc8, 0xb1, 0xbe, 0x1b, 0x39,
+	0xd6, 0xcf, 0x23, 0xc7, 0x3a, 0x1c, 0x39, 0xd6, 0x6f, 0x23, 0xc7, 0xfa, 0x7d, 0xe4, 0xe4, 0x5e,
+	0x8c, 0x1c, 0xeb, 0xd9, 0xb1, 0x93, 0x3b, 0x3c, 0x76, 0x72, 0xcf, 0x8f, 0x9d, 0xdc, 0xe7, 0xee,
+	0x90, 0x9f, 0x64, 0x41, 0xf8, 0x05, 0xff, 0x10, 0xee, 0x65, 0xbf, 0x77, 0x4b, 0xba, 0xfb, 0x77,
+	0xfe, 0x0c, 0x00, 0x00, 0xff, 0xff, 0x25, 0x91, 0x52, 0xd4, 0x54, 0x0c, 0x00, 0x00,
 }
 
 func (this *BufferedStart) Equal(that interface{}) bool {
@@ -1236,6 +1246,9 @@ func (this *StartWorkflowRequest) Equal(that interface{}) bool {
 	if !this.ContinuedFailure.Equal(that1.ContinuedFailure) {
 		return false
 	}
+	if this.CompletedRateLimitSleep != that1.CompletedRateLimitSleep {
+		return false
+	}
 	return true
 }
 func (this *StartWorkflowResponse) Equal(that interface{}) bool {
@@ -1469,7 +1482,7 @@ func (this *StartWorkflowRequest) GoString() string {
 	if this == nil {
 		return "nil"
 	}
-	s := make([]string, 0, 7)
+	s := make([]string, 0, 8)
 	s = append(s, "&schedule.StartWorkflowRequest{")
 	if this.Request != nil {
 		s = append(s, "Request: "+fmt.Sprintf("%#v", this.Request)+",\n")
@@ -1480,6 +1493,7 @@ func (this *StartWorkflowRequest) GoString() string {
 	if this.ContinuedFailure != nil {
 		s = append(s, "ContinuedFailure: "+fmt.Sprintf("%#v", this.ContinuedFailure)+",\n")
 	}
+	s = append(s, "CompletedRateLimitSleep: "+fmt.Sprintf("%#v", this.CompletedRateLimitSleep)+",\n")
 	s = append(s, "}")
 	return strings.Join(s, "")
 }
@@ -2011,6 +2025,16 @@ func (m *StartWorkflowRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.CompletedRateLimitSleep {
+		i--
+		if m.CompletedRateLimitSleep {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x30
+	}
 	if m.ContinuedFailure != nil {
 		{
 			size, err := m.ContinuedFailure.MarshalToSizedBuffer(dAtA[:i])
@@ -2419,6 +2443,9 @@ func (m *StartWorkflowRequest) Size() (n int) {
 		l = m.ContinuedFailure.Size()
 		n += 1 + l + sovMessage(uint64(l))
 	}
+	if m.CompletedRateLimitSleep {
+		n += 2
+	}
 	return n
 }
 
@@ -2618,6 +2645,7 @@ func (this *StartWorkflowRequest) String() string {
 		`Request:` + strings.Replace(fmt.Sprintf("%v", this.Request), "StartWorkflowExecutionRequest", "v14.StartWorkflowExecutionRequest", 1) + `,`,
 		`LastCompletionResult:` + strings.Replace(fmt.Sprintf("%v", this.LastCompletionResult), "Payloads", "v11.Payloads", 1) + `,`,
 		`ContinuedFailure:` + strings.Replace(fmt.Sprintf("%v", this.ContinuedFailure), "Failure", "v12.Failure", 1) + `,`,
+		`CompletedRateLimitSleep:` + fmt.Sprintf("%v", this.CompletedRateLimitSleep) + `,`,
 		`}`,
 	}, "")
 	return s
@@ -4030,6 +4058,26 @@ func (m *StartWorkflowRequest) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 6:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field CompletedRateLimitSleep", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMessage
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.CompletedRateLimitSleep = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := skipMessage(dAtA[iNdEx:])
diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go
index c054919a2c7..161a1aaae14 100644
--- a/common/dynamicconfig/constants.go
+++ b/common/dynamicconfig/constants.go
@@ -720,4 +720,6 @@ const (
 	// WorkerStickyCacheSize controls the sticky cache size for SDK workers on worker nodes
 	// (shared between all workers in the process, cannot be changed after startup)
 	WorkerStickyCacheSize = "worker.stickyCacheSize"
+	// SchedulerNamespaceStartWorkflowRPS is the per-namespace limit for starting workflows by schedules
+	SchedulerNamespaceStartWorkflowRPS = "worker.schedulerNamespaceStartWorkflowRPS"
 )
diff --git a/proto/internal/temporal/server/api/schedule/v1/message.proto b/proto/internal/temporal/server/api/schedule/v1/message.proto
index 66e01bc375b..641ae64e5dc 100644
--- a/proto/internal/temporal/server/api/schedule/v1/message.proto
+++ b/proto/internal/temporal/server/api/schedule/v1/message.proto
@@ -104,6 +104,7 @@ message StartWorkflowRequest {
     reserved 3;
     temporal.api.common.v1.Payloads last_completion_result = 4;
     temporal.api.failure.v1.Failure continued_failure = 5;
+    bool completed_rate_limit_sleep = 6;
 }
 
 message StartWorkflowResponse {
diff --git a/service/worker/batcher/fx.go b/service/worker/batcher/fx.go
index 93e532abcee..d01a7da7ae3 100644
--- a/service/worker/batcher/fx.go
+++ b/service/worker/batcher/fx.go
@@ -94,7 +94,7 @@ func (s *workerComponent) DedicatedWorkerOptions(ns *namespace.Namespace) *worke
 	}
 }
 
-func (s *workerComponent) Register(worker sdkworker.Worker, ns *namespace.Namespace) {
+func (s *workerComponent) Register(worker sdkworker.Worker, ns *namespace.Namespace, _ workercommon.RegistrationDetails) {
 	worker.RegisterWorkflowWithOptions(BatchWorkflow, workflow.RegisterOptions{Name: BatchWFTypeName})
 	worker.RegisterActivity(s.activities(ns.Name(), ns.ID()))
 }
diff --git a/service/worker/common/interface.go b/service/worker/common/interface.go
index c4863afdb5d..3a5173fdb00 100644
--- a/service/worker/common/interface.go
+++ b/service/worker/common/interface.go
@@ -54,7 +54,7 @@ type (
 	PerNSWorkerComponent interface {
 		// Register registers Workflow and Activity types provided by this worker component.
 		// The namespace that this worker is running in is also provided.
-		Register(sdkworker.Worker, *namespace.Namespace)
+		Register(sdkworker.Worker, *namespace.Namespace, RegistrationDetails)
 		// DedicatedWorkerOptions returns a PerNSDedicatedWorkerOptions for this worker component.
 		DedicatedWorkerOptions(*namespace.Namespace) *PerNSDedicatedWorkerOptions
 	}
@@ -63,4 +63,14 @@ type (
 		// Set this to false to disable this worker for this namespace
 		Enabled bool
 	}
+
+	RegistrationDetails struct {
+		// TotalWorkers is the number of requested per-namespace workers for this namespace.
+		TotalWorkers int
+		// Multiplicity is the number of those workers that this particular sdkworker.Worker
+		// represents. It may be more than one if the requested number is more than the total
+		// number of worker nodes or if consistent hashing decided to place more than one on
+		// the same node.
+		Multiplicity int
+	}
 )
diff --git a/service/worker/common/interface_mock.go b/service/worker/common/interface_mock.go
index 13ed03f329c..f89d4367d4a 100644
--- a/service/worker/common/interface_mock.go
+++ b/service/worker/common/interface_mock.go
@@ -123,13 +123,13 @@ func (mr *MockPerNSWorkerComponentMockRecorder) DedicatedWorkerOptions(arg0 inte
 }
 
 // Register mocks base method.
-func (m *MockPerNSWorkerComponent) Register(arg0 worker.Worker, arg1 *namespace.Namespace) {
+func (m *MockPerNSWorkerComponent) Register(arg0 worker.Worker, arg1 *namespace.Namespace, arg2 RegistrationDetails) {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Register", arg0, arg1)
+	m.ctrl.Call(m, "Register", arg0, arg1, arg2)
 }
 
 // Register indicates an expected call of Register.
-func (mr *MockPerNSWorkerComponentMockRecorder) Register(arg0, arg1 interface{}) *gomock.Call {
+func (mr *MockPerNSWorkerComponentMockRecorder) Register(arg0, arg1, arg2 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Register", reflect.TypeOf((*MockPerNSWorkerComponent)(nil).Register), arg0, arg1)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Register", reflect.TypeOf((*MockPerNSWorkerComponent)(nil).Register), arg0, arg1, arg2)
 }
diff --git a/service/worker/pernamespaceworker.go b/service/worker/pernamespaceworker.go
index bc975b86ef3..ae06b7edd26 100644
--- a/service/worker/pernamespaceworker.go
+++ b/service/worker/pernamespaceworker.go
@@ -246,24 +246,24 @@ func (wm *perNamespaceWorkerManager) removeWorker(ns *namespace.Namespace) {
 	delete(wm.workers, ns.ID())
 }
 
-func (wm *perNamespaceWorkerManager) getWorkerMultiplicity(ns *namespace.Namespace) (int, error) {
-	workerCount := wm.config.PerNamespaceWorkerCount(ns.Name().String())
-	// This can result in fewer than the intended number of workers if numWorkers > 1, because
+func (wm *perNamespaceWorkerManager) getWorkerMultiplicity(ns *namespace.Namespace) (int, int, error) {
+	totalWorkers := wm.config.PerNamespaceWorkerCount(ns.Name().String())
+	// This can result in fewer than the intended number of workers if totalWorkers > 1, because
 	// multiple lookups might land on the same node. To compensate, we increase the number of
 	// pollers in that case, but it would be better to try to spread them across our nodes.
 	// TODO: implement this properly using LookupN in serviceResolver
 	multiplicity := 0
-	for i := 0; i < workerCount; i++ {
+	for i := 0; i < totalWorkers; i++ {
 		key := fmt.Sprintf("%s/%d", ns.ID().String(), i)
 		target, err := wm.serviceResolver.Lookup(key)
 		if err != nil {
-			return 0, err
+			return 0, 0, err
 		}
 		if target.Identity() == wm.self.Identity() {
 			multiplicity++
 		}
 	}
-	return multiplicity, nil
+	return multiplicity, totalWorkers, nil
 }
 
 func (wm *perNamespaceWorkerManager) getWorkerOptions(ns *namespace.Namespace) sdkWorkerOptions {
@@ -379,7 +379,7 @@ func (w *perNamespaceWorker) tryRefresh(ns *namespace.Namespace) error {
 	}
 
 	// check if we are responsible for this namespace at all
-	multiplicity, err := w.wm.getWorkerMultiplicity(ns)
+	multiplicity, totalWorkers, err := w.wm.getWorkerMultiplicity(ns)
 	if err != nil {
 		w.logger.Error("Failed to look up hosts", tag.Error(err))
 		// TODO: add metric also
@@ -410,7 +410,7 @@ func (w *perNamespaceWorker) tryRefresh(ns *namespace.Namespace) error {
 	// create new one. note that even before startWorker returns, the worker may have started
 	// and already called the fatal error handler. we need to set w.client+worker+componentSet
 	// before releasing the lock to keep our state consistent.
-	client, worker, err := w.startWorker(ns, enabledComponents, multiplicity, dcOptions)
+	client, worker, err := w.startWorker(ns, enabledComponents, multiplicity, totalWorkers, dcOptions)
 	if err != nil {
 		// TODO: add metric also
 		return err
@@ -426,6 +426,7 @@ func (w *perNamespaceWorker) startWorker(
 	ns *namespace.Namespace,
 	components []workercommon.PerNSWorkerComponent,
 	multiplicity int,
+	totalWorkers int,
 	dcOptions sdkWorkerOptions,
 ) (sdkclient.Client, sdkworker.Worker, error) {
 	nsName := ns.Name().String()
@@ -456,8 +457,12 @@ func (w *perNamespaceWorker) startWorker(
 
 	// this should not block because the client already has server capabilities
 	worker := w.wm.sdkClientFactory.NewWorker(client, primitives.PerNSWorkerTaskQueue, sdkoptions)
+	details := workercommon.RegistrationDetails{
+		TotalWorkers: totalWorkers,
+		Multiplicity: multiplicity,
+	}
 	for _, cmp := range components {
-		cmp.Register(worker, ns)
+		cmp.Register(worker, ns, details)
 	}
 
 	// this blocks by calling DescribeNamespace a few times (with a 10s timeout)
diff --git a/service/worker/pernamespaceworker_test.go b/service/worker/pernamespaceworker_test.go
index 31b104fd758..61a71365e97 100644
--- a/service/worker/pernamespaceworker_test.go
+++ b/service/worker/pernamespaceworker_test.go
@@ -182,7 +182,7 @@ func (s *perNsWorkerManagerSuite) TestEnabled() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
 	wkr1 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
-	s.cmp1.EXPECT().Register(wkr1, ns)
+	s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 	wkr1.EXPECT().Start()
 
 	s.manager.namespaceCallback(ns, false)
@@ -213,7 +213,7 @@ func (s *perNsWorkerManagerSuite) TestMultiplicity() {
 		s.Equal(4, options.MaxConcurrentWorkflowTaskPollers)
 		s.Equal(4, options.MaxConcurrentActivityTaskPollers)
 	}).Return(wkr1)
-	s.cmp1.EXPECT().Register(wkr1, ns)
+	s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 3, Multiplicity: 2})
 	wkr1.EXPECT().Start()
 
 	s.manager.namespaceCallback(ns, false)
@@ -258,7 +258,7 @@ func (s *perNsWorkerManagerSuite) TestOptions() {
 		s.Equal(0.0, options.WorkerLocalActivitiesPerSecond)
 		s.Equal(0*time.Millisecond, options.StickyScheduleToStartTimeout)
 	}).Return(wkr)
-	s.cmp1.EXPECT().Register(wkr, gomock.Any()).AnyTimes()
+	s.cmp1.EXPECT().Register(wkr, gomock.Any(), gomock.Any()).AnyTimes()
 	wkr.EXPECT().Start().AnyTimes()
 
 	s.manager.namespaceCallback(ns1, false)
@@ -302,9 +302,9 @@ func (s *perNsWorkerManagerSuite) TestTwoNamespacesTwoComponents() {
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr2)
 
-	s.cmp1.EXPECT().Register(wkr1, ns1)
-	s.cmp1.EXPECT().Register(wkr2, ns2)
-	s.cmp2.EXPECT().Register(wkr1, ns1)
+	s.cmp1.EXPECT().Register(wkr1, ns1, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
+	s.cmp1.EXPECT().Register(wkr2, ns2, workercommon.RegistrationDetails{TotalWorkers: 2, Multiplicity: 2})
+	s.cmp2.EXPECT().Register(wkr1, ns1, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 
 	wkr1.EXPECT().Start()
 	wkr2.EXPECT().Start()
@@ -334,7 +334,7 @@ func (s *perNsWorkerManagerSuite) TestDeleteNs() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
 	wkr1 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
-	s.cmp1.EXPECT().Register(wkr1, ns)
+	s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 	wkr1.EXPECT().Start()
 
 	s.manager.namespaceCallback(ns, false)
@@ -354,7 +354,7 @@ func (s *perNsWorkerManagerSuite) TestDeleteNs() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli2)
 	wkr2 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr2)
-	s.cmp1.EXPECT().Register(wkr2, ns)
+	s.cmp1.EXPECT().Register(wkr2, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 	wkr2.EXPECT().Start()
 
 	s.manager.namespaceCallback(nsRestored, false)
@@ -389,7 +389,7 @@ func (s *perNsWorkerManagerSuite) TestMembershipChanged() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
 	wkr1 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
-	s.cmp1.EXPECT().Register(wkr1, ns)
+	s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 	wkr1.EXPECT().Start()
 
 	s.manager.membershipChangedCh <- nil
@@ -422,7 +422,7 @@ func (s *perNsWorkerManagerSuite) TestServiceResolverError() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
 	wkr1 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
-	s.cmp1.EXPECT().Register(wkr1, ns)
+	s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 	wkr1.EXPECT().Start()
 
 	s.manager.namespaceCallback(ns, false)
@@ -449,7 +449,7 @@ func (s *perNsWorkerManagerSuite) TestStartWorkerError() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli1)
 	wkr1 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli1}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr1)
-	s.cmp1.EXPECT().Register(wkr1, ns)
+	s.cmp1.EXPECT().Register(wkr1, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 
 	// first try fails to start
 	wkr1.EXPECT().Start().Return(errors.New("start worker error"))
@@ -460,7 +460,7 @@ func (s *perNsWorkerManagerSuite) TestStartWorkerError() {
 	s.cfactory.EXPECT().NewClient(matchOptions("ns1")).Return(cli2)
 	wkr2 := mocksdk.NewMockWorker(s.controller)
 	s.cfactory.EXPECT().NewWorker(matchStrict{cli2}, primitives.PerNSWorkerTaskQueue, gomock.Any()).Return(wkr2)
-	s.cmp1.EXPECT().Register(wkr2, ns)
+	s.cmp1.EXPECT().Register(wkr2, ns, workercommon.RegistrationDetails{TotalWorkers: 1, Multiplicity: 1})
 	wkr2.EXPECT().Start()
 
 	s.manager.namespaceCallback(ns, false)
diff --git a/service/worker/scheduler/activities.go b/service/worker/scheduler/activities.go
index ba608defa78..58530fad678 100644
--- a/service/worker/scheduler/activities.go
+++ b/service/worker/scheduler/activities.go
@@ -44,6 +44,7 @@ import (
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/primitives/timestamp"
+	"go.temporal.io/server/common/quotas"
 )
 
 type (
@@ -51,9 +52,16 @@ type (
 		activityDeps
 		namespace   namespace.Name
 		namespaceID namespace.ID
+		// Rate limiter for start workflow requests. Note that the scope is all schedules in
+		// this namespace on this worker.
+		startWorkflowRateLimiter quotas.RateLimiter
 	}
 
 	errFollow string
+
+	rateLimitedDetails struct {
+		Delay time.Duration
+	}
 )
 
 var (
@@ -61,11 +69,27 @@ var (
 	errWrongChain = errors.New("found running workflow with wrong FirstExecutionRunId")
 	errNoEvents   = errors.New("GetEvents didn't return any events")
 	errNoAttrs    = errors.New("last event did not have correct attrs")
+	errBlocked    = errors.New("rate limiter doesn't allow any progress")
 )
 
 func (e errFollow) Error() string { return string(e) }
 
 func (a *activities) StartWorkflow(ctx context.Context, req *schedspb.StartWorkflowRequest) (*schedspb.StartWorkflowResponse, error) {
+	if !req.CompletedRateLimitSleep {
+		reservation := a.startWorkflowRateLimiter.Reserve()
+		if !reservation.OK() {
+			return nil, translateError(errBlocked, "StartWorkflowExecution")
+		}
+		delay := reservation.Delay()
+		if delay > 1*time.Second {
+			// for a long sleep, ask the workflow to do it in workflow logic
+			return nil, temporal.NewNonRetryableApplicationError(
+				rateLimitedErrorType, rateLimitedErrorType, nil, rateLimitedDetails{Delay: delay})
+		}
+		// short sleep can be done in-line
+		time.Sleep(delay)
+	}
+
 	req.Request.Namespace = a.namespace.String()
 
 	request := common.CreateHistoryStartWorkflowRequest(
diff --git a/service/worker/scheduler/fx.go b/service/worker/scheduler/fx.go
index 70deda320e5..3b7414f7c06 100644
--- a/service/worker/scheduler/fx.go
+++ b/service/worker/scheduler/fx.go
@@ -36,6 +36,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/quotas"
 	workercommon "go.temporal.io/server/service/worker/common"
 )
 
@@ -46,8 +47,9 @@ const (
 
 type (
 	workerComponent struct {
-		activityDeps activityDeps
-		enabledForNs dynamicconfig.BoolPropertyFnWithNamespaceFilter
+		activityDeps             activityDeps
+		enabledForNs             dynamicconfig.BoolPropertyFnWithNamespaceFilter
+		globalNSStartWorkflowRPS dynamicconfig.FloatPropertyFnWithNamespaceFilter
 	}
 
 	activityDeps struct {
@@ -77,6 +79,8 @@ func NewResult(
 			activityDeps: params,
 			enabledForNs: dcCollection.GetBoolPropertyFnWithNamespaceFilter(
 				dynamicconfig.WorkerEnableScheduler, true),
+			globalNSStartWorkflowRPS: dcCollection.GetFloatPropertyFilteredByNamespace(
+				dynamicconfig.SchedulerNamespaceStartWorkflowRPS, 30.0),
 		},
 	}
 }
@@ -87,15 +91,19 @@ func (s *workerComponent) DedicatedWorkerOptions(ns *namespace.Namespace) *worke
 	}
 }
 
-func (s *workerComponent) Register(worker sdkworker.Worker, ns *namespace.Namespace) {
+func (s *workerComponent) Register(worker sdkworker.Worker, ns *namespace.Namespace, details workercommon.RegistrationDetails) {
 	worker.RegisterWorkflowWithOptions(SchedulerWorkflow, workflow.RegisterOptions{Name: WorkflowType})
-	worker.RegisterActivity(s.activities(ns.Name(), ns.ID()))
+	worker.RegisterActivity(s.activities(ns.Name(), ns.ID(), details))
 }
 
-func (s *workerComponent) activities(name namespace.Name, id namespace.ID) *activities {
+func (s *workerComponent) activities(name namespace.Name, id namespace.ID, details workercommon.RegistrationDetails) *activities {
+	localRPS := func() float64 {
+		return float64(details.Multiplicity) * s.globalNSStartWorkflowRPS(name.String()) / float64(details.TotalWorkers)
+	}
 	return &activities{
-		activityDeps: s.activityDeps,
-		namespace:    name,
-		namespaceID:  id,
+		activityDeps:             s.activityDeps,
+		namespace:                name,
+		namespaceID:              id,
+		startWorkflowRateLimiter: quotas.NewDefaultOutgoingRateLimiter(localRPS),
 	}
 }
diff --git a/service/worker/scheduler/workflow.go b/service/worker/scheduler/workflow.go
index d1d46ee9464..0d2c05bc76b 100644
--- a/service/worker/scheduler/workflow.go
+++ b/service/worker/scheduler/workflow.go
@@ -73,6 +73,8 @@ const (
 	maxListMatchingTimesCount = 1000
 
 	invalidDuration time.Duration = -1
+
+	rateLimitedErrorType = "RateLimited"
 )
 
 type (
@@ -676,6 +678,7 @@ func (s *scheduler) addStart(nominalTime, actualTime time.Time, overlapPolicy en
 }
 
 // processBuffer should return true if there might be more work to do right now.
+//
 //nolint:revive
 func (s *scheduler) processBuffer() bool {
 	s.logger.Debug("processBuffer", "buffer", len(s.State.BufferedStarts), "running", len(s.Info.RunningWorkflows), "needRefresh", s.State.NeedRefresh)
@@ -810,20 +813,32 @@ func (s *scheduler) startWorkflow(
 		LastCompletionResult: s.State.LastCompletionResult,
 		ContinuedFailure:     s.State.ContinuedFailure,
 	}
-	var res schedspb.StartWorkflowResponse
-	err := workflow.ExecuteLocalActivity(ctx, s.a.StartWorkflow, req).Get(s.ctx, &res)
-	if err != nil {
-		return nil, err
-	}
+	for {
+		var res schedspb.StartWorkflowResponse
+		err := workflow.ExecuteLocalActivity(ctx, s.a.StartWorkflow, req).Get(s.ctx, &res)
+
+		var appErr *temporal.ApplicationError
+		var details rateLimitedDetails
+		if errors.As(err, &appErr) &&
+			appErr.Type() == rateLimitedErrorType &&
+			appErr.Details(&details) == nil {
+			workflow.Sleep(s.ctx, details.Delay)
+			req.CompletedRateLimitSleep = true // only use rate limiter once
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
 
-	return &schedpb.ScheduleActionResult{
-		ScheduleTime: start.ActualTime,
-		ActualTime:   res.RealStartTime,
-		StartWorkflowResult: &commonpb.WorkflowExecution{
-			WorkflowId: workflowID,
-			RunId:      res.RunId,
-		},
-	}, nil
+		return &schedpb.ScheduleActionResult{
+			ScheduleTime: start.ActualTime,
+			ActualTime:   res.RealStartTime,
+			StartWorkflowResult: &commonpb.WorkflowExecution{
+				WorkflowId: workflowID,
+				RunId:      res.RunId,
+			},
+		}, nil
+	}
 }
 
 func (s *scheduler) identity() string {
diff --git a/tests/schedule_test.go b/tests/schedule_test.go
index 5eed6467935..17013f4f891 100644
--- a/tests/schedule_test.go
+++ b/tests/schedule_test.go
@@ -46,6 +46,7 @@ import (
 	"go.temporal.io/sdk/worker"
 	"go.temporal.io/sdk/workflow"
 
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/payload"
 	"go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite"
@@ -547,3 +548,82 @@ func (s *scheduleIntegrationSuite) TestRefresh() {
 	events3 := s.getHistory(s.namespace, &commonpb.WorkflowExecution{WorkflowId: scheduler.WorkflowIDPrefix + sid})
 	s.Greater(len(events3), len(events2))
 }
+
+func (s *scheduleIntegrationSuite) TestRateLimit() {
+	sid := "sched-test-rate-limit-%d"
+	wid := "sched-test-rate-limit-wf-%d"
+	wt := "sched-test-rate-limit-wt"
+
+	// Set 1/sec rate limit per namespace. To force this to take effect immediately (instead of
+	// waiting one minute) we have to cause the whole worker to be stopped and started. The
+	// sleeps are needed because the refresh is asynchronous, and there's no way to get access
+	// to the actual rate limiter object to refresh it directly.
+	s.testCluster.host.dcClient.OverrideValue(dynamicconfig.SchedulerNamespaceStartWorkflowRPS, 1.0)
+	s.testCluster.host.dcClient.OverrideValue(dynamicconfig.WorkerPerNamespaceWorkerCount, 0)
+	s.testCluster.host.workerService.RefreshPerNSWorkerManager()
+	time.Sleep(2 * time.Second)
+	s.testCluster.host.dcClient.RemoveOverride(dynamicconfig.WorkerPerNamespaceWorkerCount)
+	s.testCluster.host.workerService.RefreshPerNSWorkerManager()
+	time.Sleep(2 * time.Second)
+
+	var runs int32
+	workflowFn := func(ctx workflow.Context) error {
+		workflow.SideEffect(ctx, func(ctx workflow.Context) any {
+			atomic.AddInt32(&runs, 1)
+			return 0
+		})
+		return nil
+	}
+	s.worker.RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
+
+	// create 10 copies of the schedule
+	for i := 0; i < 10; i++ {
+		schedule := &schedulepb.Schedule{
+			Spec: &schedulepb.ScheduleSpec{
+				Interval: []*schedulepb.IntervalSpec{
+					{Interval: timestamp.DurationPtr(1 * time.Second)},
+				},
+			},
+			Action: &schedulepb.ScheduleAction{
+				Action: &schedulepb.ScheduleAction_StartWorkflow{
+					StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
+						WorkflowId:   fmt.Sprintf(wid, i),
+						WorkflowType: &commonpb.WorkflowType{Name: wt},
+						TaskQueue:    &taskqueuepb.TaskQueue{Name: s.taskQueue},
+					},
+				},
+			},
+		}
+		_, err := s.engine.CreateSchedule(NewContext(), &workflowservice.CreateScheduleRequest{
+			Namespace:  s.namespace,
+			ScheduleId: fmt.Sprintf(sid, i),
+			Schedule:   schedule,
+			Identity:   "test",
+			RequestId:  uuid.New(),
+		})
+		s.NoError(err)
+	}
+
+	time.Sleep(5 * time.Second)
+
+	// With no rate limit, we'd see 10/second == 50 workflows run. With a limit of 1/sec, we
+	// expect to see around 5.
+	s.Less(atomic.LoadInt32(&runs), int32(10))
+
+	// clean up
+	for i := 0; i < 10; i++ {
+		_, err := s.engine.DeleteSchedule(NewContext(), &workflowservice.DeleteScheduleRequest{
+			Namespace:  s.namespace,
+			ScheduleId: fmt.Sprintf(sid, i),
+			Identity:   "test",
+		})
+		s.NoError(err)
+	}
+
+	s.testCluster.host.dcClient.RemoveOverride(dynamicconfig.SchedulerNamespaceStartWorkflowRPS)
+	s.testCluster.host.dcClient.OverrideValue(dynamicconfig.WorkerPerNamespaceWorkerCount, 0)
+	s.testCluster.host.workerService.RefreshPerNSWorkerManager()
+	time.Sleep(2 * time.Second)
+	s.testCluster.host.dcClient.RemoveOverride(dynamicconfig.WorkerPerNamespaceWorkerCount)
+	s.testCluster.host.workerService.RefreshPerNSWorkerManager()
+}