[logdog] Replace Tumble with push queues.

logdog/appengine/coordinator/endpoints/services/registerStream.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 package services
 
 import (
         "crypto/subtle"
+        "time"
 
-        "github.com/golang/protobuf/proto"
-        ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/common/clock"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/grpc/grpcutil"
         "github.com/luci/luci-go/logdog/api/endpoints/coordinator/services/v1"
         "github.com/luci/luci-go/logdog/api/logpb"
         "github.com/luci/luci-go/logdog/appengine/coordinator"
         "github.com/luci/luci-go/logdog/appengine/coordinator/endpoints"
-        "github.com/luci/luci-go/logdog/appengine/coordinator/mutations"
+        "github.com/luci/luci-go/logdog/appengine/coordinator/tasks"
         "github.com/luci/luci-go/logdog/common/types"
-        "github.com/luci/luci-go/tumble"
+
+        ds "github.com/luci/gae/service/datastore"
+
+        "github.com/golang/protobuf/proto"
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 )
 
 func buildLogStreamState(ls *coordinator.LogStream, lst *coordinator.LogStreamState) *logdog.LogStreamState {
         st := logdog.LogStreamState{
                 ProtoVersion:  ls.ProtoVersion,
                 Secret:        lst.Secret,
                 TerminalIndex: lst.TerminalIndex,
                 Archived:      lst.ArchivalState().Archived(),
@@ skipping 104 matching lines @@
         ls := &coordinator.LogStream{ID: logStreamID}
         lst := ls.State(c)
 
         if err := ds.Get(c, ls, lst); err != nil {
                 if !anyNoSuchEntity(err) {
                         log.WithError(err).Errorf(c, "Failed to check for log stream.")
                         return nil, err
                 }
 
                 // The stream does not exist. Proceed with transactional registration.
-                lstKey := ds.KeyForObj(c, lst)
-                err = tumble.RunUnbuffered(c, lstKey, func(c context.Context) ([]tumble.Mutation, error) {
+                err = ds.RunInTransaction(c, func(c context.Context) error {
                         // Load our state and stream (transactional).
                         switch err := ds.Get(c, ls, lst); {
                         case err == nil:
                                 // The stream is already registered.
-                                return nil, nil
+                                return nil
 
                         case !anyNoSuchEntity(err):
                                 log.WithError(err).Errorf(c, "Failed to check for stream registration (transactional).")
-                                return nil, err
+                                return err
                         }
 
                         // The stream is not yet registered.
                         log.Infof(c, "Registering new log stream.")
 
                         // Construct our LogStreamState.
                         now := clock.Now(c).UTC()
                         lst.Created = now
                         lst.Updated = now
                         lst.Secret = pfx.Secret // Copy Prefix Secret to reduce datastore Gets.
 
                         // Construct our LogStream.
                         ls.Created = now
                         ls.ProtoVersion = req.ProtoVersion
 
                         if err := ls.LoadDescriptor(&desc); err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                 }.Errorf(c, "Failed to load descriptor into LogStream.")
-                                return nil, grpcutil.Errf(codes.InvalidArgument, "Failed to load descriptor.")
+                                return grpcutil.Errf(codes.InvalidArgument, "Failed to load descriptor.")
                         }
 
                         // If our registration request included a terminal index, terminate the
                         // log stream state as well.
                         if req.TerminalIndex >= 0 {
                                 log.Fields{
                                         "terminalIndex": req.TerminalIndex,
                                 }.Debugf(c, "Registration request included terminal index.")
 
                                 lst.TerminalIndex = req.TerminalIndex
                                 lst.TerminatedTime = now
                         } else {
                                 lst.TerminalIndex = -1
                         }
 
                         if err := ds.Put(c, ls, lst); err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                 }.Errorf(c, "Failed to Put LogStream.")
-                                return nil, grpcutil.Internal
+                                return grpcutil.Internal
                         }
 
-                        // Add a named delayed mutation to archive this stream if it's not archived
-                        // yet.
+                        // Add a named delayed task queue task to archive this stream if it's not
+                        // archived yet.
                         //
                         // If the registration did not include a terminal index, this will be our
-                        // pessimistic archival request, scheduled on registration to catch streams
-                        // that don't expire. This mutation will be replaced by the optimistic
-                        // archival mutation when/if the stream is terminated via TerminateStream.
+                        // pessimistic archival request, scheduled on registration to catch
+                        // streams that don't expire. This task will be replaced by the optimistic
+                        // archival task when/if the stream is terminated via TerminateStream.
                         //
                         // If the registration included a terminal index, apply our standard
                         // parameters to the archival. Since TerminateStream will not be called,
                         // this will be our formal optimistic archival task.
                         params := standardArchivalParams(cfg, pcfg)
-                        cat := mutations.CreateArchiveTask{
-                                ID: ls.ID,
-                        }
+
+                        var (
+                                delay       time.Duration
+                                archivalTag logdog.ArchiveDispatchTask_Tag
+                        )
                         if req.TerminalIndex < 0 {
                                 // No terminal index, schedule pessimistic cleanup archival.
-                                cat.Expiration = now.Add(params.CompletePeriod)
+                                delay = params.CompletePeriod
+                                archivalTag = logdog.ArchiveDispatchTask_EXPIRED
+
+                                // For cleanup, we instruct the archival to not wait any longer or
+                                // allow the stream time extra time to become complete (archive as-is).
+                                params.SettleDelay = 0
+                                params.CompletePeriod = 0
 
                                 log.Fields{
-                                        "deadline": cat.Expiration,
-                                }.Debugf(c, "Scheduling cleanup archival mutation.")
+                                        "deadline": delay,
+                                }.Debugf(c, "Scheduling cleanup archival task.")
                         } else {
-                                // Terminal index, schedule optimistic archival (mirrors TerminateStream).
-                                cat.SettleDelay = params.SettleDelay
-                                cat.CompletePeriod = params.CompletePeriod
-
-                                // Schedule this mutation to execute after our settle delay.
-                                cat.Expiration = now.Add(params.SettleDelay)
+                                // Schedule this task to execute after our settle delay.
+                                delay = params.SettleDelay
+                                archivalTag = logdog.ArchiveDispatchTask_TERMINATED
 
                                 log.Fields{
-                                        "settleDelay":    cat.SettleDelay,
-                                        "completePeriod": cat.CompletePeriod,
-                                        "scheduledAt":    cat.Expiration,
-                                }.Debugf(c, "Scheduling archival mutation.")
+                                        "settleDelay":    params.SettleDelay,
+                                        "completePeriod": params.CompletePeriod,
+                                        "scheduledAt":    delay,
+                                }.Debugf(c, "Scheduling archival task.")
                         }
 
-                        aeName := cat.TaskName(c)
-                        if err := tumble.PutNamedMutations(c, lstKey, map[string]tumble.Mutation{aeName: &cat}); err != nil {
-                                log.WithError(err).Errorf(c, "Failed to write named mutations.")
-                                return nil, grpcutil.Internal
+                        if err := tasks.CreateArchivalTask(c, logStreamID, archivalTag, delay, params); err != nil {
+                                log.WithError(err).Errorf(c, "Failed to create archival task.")
+                                return grpcutil.Internal
                         }
 
-                        return nil, nil
-                })
+                        return nil
+                }, nil)
                 if err != nil {
                         log.Fields{
                                 log.ErrorKey: err,
                         }.Errorf(c, "Failed to register LogStream.")
                         return nil, err
                 }
         }
 
         return &logdog.RegisterStreamResponse{
                 Id:    string(ls.ID),
                 State: buildLogStreamState(ls, lst),
         }, nil
 }