feat: grade open answers through a grader registry in the background
- graders registry collected at boot from every HasGraders plugin, manual first; a duplicate or reserved code fails boot - grading job on queue quizzes_grading with three attempts; a stale or reviewed answer is never touched; the last failure marks it failed - learning answers enqueue at once, tests at close; a grade recalculates a finished attempt and fires AnswerGradedEvent after commit
This commit is contained in:
Jakub Zych
373
classes/grading.go
Normal file
373
classes/grading.go
Normal file
@@ -0,0 +1,373 @@
|
||||
package classes
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.golem15.com/golem15/sm-quizzes-plugin/events"
|
||||
"git.golem15.com/golem15/sm-quizzes-plugin/graders"
|
||||
"git.golem15.com/golem15/sm-quizzes-plugin/jobs"
|
||||
"git.golem15.com/golem15/sm-quizzes-plugin/models"
|
||||
"git.golem15.com/golem15/summercms/modules/conga"
|
||||
"git.golem15.com/golem15/summercms/modules/lagoon"
|
||||
"git.golem15.com/golem15/summercms/modules/pact"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
// GradeMaxAttempts is how many times one open answer is handed to its
|
||||
// grader before it is marked failed (D-17). The grading job's River attempt
|
||||
// limit is the same number.
|
||||
const GradeMaxAttempts = 3
|
||||
|
||||
// GradeTimeout is the deadline of one grading job attempt.
|
||||
const GradeTimeout = 3 * time.Minute
|
||||
|
||||
// Text limits of what a grading writes.
|
||||
const (
|
||||
maxGraderFeedback = 2000
|
||||
maxGradingError = 500
|
||||
)
|
||||
|
||||
// errInvalidScore is a grader result ClampOpenPoints refuses; it counts as
|
||||
// a failed attempt.
|
||||
var errInvalidScore = errors.New("quizzes: grader returned an invalid score")
|
||||
|
||||
// Enqueuer inserts a background job, joining db's transaction when there is
|
||||
// one. *conga.Manager implements it; tests record the jobs instead.
|
||||
type Enqueuer interface {
|
||||
Enqueue(ctx context.Context, db *gorm.DB, args pact.JobArgs, o conga.EnqueueOpts) error
|
||||
}
|
||||
|
||||
// GraderFor returns the grader of a quiz: false for a nil registry or quiz,
|
||||
// for manual and for a code no activated plugin registered (D-05). An
|
||||
// answer without a grader waits for the owner.
|
||||
func GraderFor(reg *graders.Registry, quiz *models.Quiz) (graders.Grader, bool) {
|
||||
if quiz == nil {
|
||||
return nil, false
|
||||
}
|
||||
return reg.Get(quiz.Grader)
|
||||
}
|
||||
|
||||
// GradingStep is what a grading job does with an answer.
|
||||
type GradingStep int
|
||||
|
||||
// Grading steps.
|
||||
const (
|
||||
// GradingSkip leaves the answer alone: it is no longer pending or it
|
||||
// has a newer text than the job.
|
||||
GradingSkip GradingStep = iota
|
||||
// GradingRetry returns the grader's error so River retries the job.
|
||||
GradingRetry
|
||||
// GradingFail marks the answer failed after the last attempt.
|
||||
GradingFail
|
||||
// GradingStore stores the grader's score.
|
||||
GradingStore
|
||||
)
|
||||
|
||||
// NextGrading decides a grading job's step from the answer's status and
|
||||
// grading_seq, the job's seq, the attempts used for this text (the current
|
||||
// one included) and the grader's error. It is pure.
|
||||
func NextGrading(status *string, rowSeq, jobSeq, attempts int, gradeErr error) GradingStep {
|
||||
if status == nil || *status != models.GradingPending || rowSeq != jobSeq {
|
||||
return GradingSkip
|
||||
}
|
||||
if gradeErr == nil {
|
||||
return GradingStore
|
||||
}
|
||||
if attempts >= GradeMaxAttempts {
|
||||
return GradingFail
|
||||
}
|
||||
return GradingRetry
|
||||
}
|
||||
|
||||
// enqueueGrading hands a pending open answer to the quiz's grader as a
|
||||
// background job inside tx, so the job exists only if the answer commits.
|
||||
// Without a job queue or a registered grader (manual, unknown) it does
|
||||
// nothing: the answer waits for the owner.
|
||||
func (s *Service) enqueueGrading(ctx context.Context, tx *gorm.DB, answerID uint, seq int, attempt *models.Attempt) error {
|
||||
if s.Jobs == nil || attempt == nil {
|
||||
return nil
|
||||
}
|
||||
if _, ok := GraderFor(s.Graders, attempt.Quiz); !ok {
|
||||
return nil
|
||||
}
|
||||
return s.Jobs.Enqueue(ctx, tx, jobs.GradeAnswerArgs{AnswerID: answerID, Seq: seq}, conga.EnqueueOpts{
|
||||
Queue: jobs.Queue,
|
||||
MaxAttempts: GradeMaxAttempts,
|
||||
})
|
||||
}
|
||||
|
||||
// enqueuePending enqueues every pending open answer of a test attempt that
|
||||
// is being closed, one job per answer (UD-2): answers given during a test
|
||||
// are graded only once it ends.
|
||||
func (s *Service) enqueuePending(ctx context.Context, tx *gorm.DB, attempt *models.Attempt) error {
|
||||
if s.Jobs == nil || !IsTest(attempt.Quiz) {
|
||||
return nil
|
||||
}
|
||||
if _, ok := GraderFor(s.Graders, attempt.Quiz); !ok {
|
||||
return nil
|
||||
}
|
||||
var rows []models.AttemptAnswer
|
||||
if err := tx.WithContext(ctx).Where("attempt_id = ? AND grading_status = ?", attempt.ID, models.GradingPending).
|
||||
Order("id ASC").Find(&rows).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
for _, r := range rows {
|
||||
if err := s.enqueueGrading(ctx, tx, r.ID, r.GradingSeq, attempt); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// lockAttempt locks the attempt row FOR UPDATE, so a grade, a review and
|
||||
// Finish of one attempt never interleave. A missing attempt is
|
||||
// gorm.ErrRecordNotFound.
|
||||
func lockAttempt(ctx context.Context, tx *gorm.DB, attemptID uint) (*models.Attempt, error) {
|
||||
var a models.Attempt
|
||||
err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE", Table: clause.Table{Name: clause.CurrentTable}}).
|
||||
Where("id = ?", attemptID).Take(&a).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &a, nil
|
||||
}
|
||||
|
||||
// recalcAttempt recomputes the score of a completed attempt from its
|
||||
// answers after a grade or a review (Pitfall 2), under the attempt row
|
||||
// lock. The percentage is taken against the attempt's stored max_points,
|
||||
// so a later question edit does not move an old attempt's maximum. An open
|
||||
// or missing attempt is left alone: close scores it.
|
||||
func (s *Service) recalcAttempt(ctx context.Context, tx *gorm.DB, attemptID uint) error {
|
||||
tx = tx.WithContext(ctx)
|
||||
a, err := lockAttempt(ctx, tx, attemptID)
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if a.CompletedAt == nil {
|
||||
return nil
|
||||
}
|
||||
var points []float64
|
||||
if err := tx.Model(&models.AttemptAnswer{}).Where("attempt_id = ?", attemptID).Pluck("points", &points).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
score, percent := Totals(points, a.MaxPoints)
|
||||
return tx.Model(&models.Attempt{}).Where("id = ?", attemptID).Updates(map[string]any{
|
||||
"score_points": score,
|
||||
"score_percent": percent,
|
||||
"updated_at": s.now().UTC(),
|
||||
}).Error
|
||||
}
|
||||
|
||||
// fireAfterCommit fires event on s.Events once tx commits (never when it
|
||||
// rolls back). A listener error is logged with the event name and never
|
||||
// returned: the write it announces stays (D-14).
|
||||
func fireAfterCommit[T any](ctx context.Context, s *Service, tx *gorm.DB, name string, event T) {
|
||||
if s == nil || s.Events == nil {
|
||||
return
|
||||
}
|
||||
bus := s.Events
|
||||
lagoon.AfterCommit(ctx, tx, func(ctx context.Context, _ *gorm.DB) {
|
||||
if err := bus.Fire(ctx, event); err != nil {
|
||||
pluginLog.Warn("quizzes: event listener failed", "event", name, "error", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// GradeAnswer is the body of the grading job: it asks the quiz's grader to
|
||||
// score one pending open answer (D-06). The grader runs with no transaction
|
||||
// open and no row lock held, so a slow grader blocks no student.
|
||||
//
|
||||
// A missing answer, one that is no longer pending (graded, failed or
|
||||
// reviewed by the owner) and one whose grading_seq differs from seq (the
|
||||
// student wrote a newer text) are left alone, as is an answer whose quiz
|
||||
// has no registered grader. Each run counts an attempt for the current
|
||||
// text first. A grader error, or a score ClampOpenPoints refuses, is
|
||||
// returned for River to retry while fewer than GradeMaxAttempts attempts
|
||||
// were made; the last one marks the answer failed with the error and
|
||||
// returns nil (D-17). A score is stored as graded with points clamped to
|
||||
// 0..question points in steps of 0.5, the attempt's score is recalculated
|
||||
// when it is completed, and AnswerGradedEvent fires after the commit.
|
||||
func (s *Service) GradeAnswer(ctx context.Context, answerID uint, seq int) error {
|
||||
db := s.db(ctx)
|
||||
var row models.AttemptAnswer
|
||||
if err := db.Where("id = ?", answerID).Take(&row).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if NextGrading(row.GradingStatus, row.GradingSeq, seq, 0, nil) == GradingSkip {
|
||||
return nil
|
||||
}
|
||||
var question models.Question
|
||||
if err := db.Where("id = ?", row.QuestionID).Take(&question).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
var attempt models.Attempt
|
||||
if err := db.Preload("Quiz").Where("id = ?", row.AttemptID).Take(&attempt).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
grader, ok := GraderFor(s.Graders, attempt.Quiz)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
var attempts int
|
||||
if err := db.Raw(`UPDATE golem15_quizzes_attempt_answers SET grading_attempts = grading_attempts + 1
|
||||
WHERE id = ? AND grading_status = ? AND grading_seq = ? RETURNING grading_attempts`,
|
||||
answerID, models.GradingPending, seq).Scan(&attempts).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if attempts == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
res, gradeErr := grader.Grade(ctx, graders.Input{
|
||||
QuizID: attempt.QuizID,
|
||||
QuestionID: question.ID,
|
||||
AttemptID: attempt.ID,
|
||||
AnswerID: row.ID,
|
||||
UserID: attempt.UserID,
|
||||
Question: question.Text,
|
||||
ModelAnswer: deref(question.AnswerText),
|
||||
Rubric: deref(question.Rubric),
|
||||
Answer: deref(row.TextAnswer),
|
||||
Points: question.Points,
|
||||
})
|
||||
var points float64
|
||||
if gradeErr == nil {
|
||||
p, ok := ClampOpenPoints(res.Points, question.Points)
|
||||
if !ok {
|
||||
gradeErr = errInvalidScore
|
||||
}
|
||||
points = p
|
||||
}
|
||||
pending := models.GradingPending
|
||||
switch NextGrading(&pending, seq, seq, attempts, gradeErr) {
|
||||
case GradingRetry:
|
||||
return gradeErr
|
||||
case GradingFail:
|
||||
return s.failGrading(ctx, row.ID, attempt.ID, seq, grader.Code(), gradeErr)
|
||||
case GradingStore:
|
||||
return s.storeGrade(ctx, row.ID, seq, &attempt, question, grader.Code(), points, res.Feedback)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// stillPending re-reads an answer inside the grading write's transaction
|
||||
// (the attempt row is locked) and reports whether the job's seq still
|
||||
// applies.
|
||||
func stillPending(ctx context.Context, tx *gorm.DB, answerID uint, seq int) (bool, error) {
|
||||
var cur models.AttemptAnswer
|
||||
err := tx.WithContext(ctx).Where("id = ?", answerID).Take(&cur).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return NextGrading(cur.GradingStatus, cur.GradingSeq, seq, 0, nil) != GradingSkip, nil
|
||||
}
|
||||
|
||||
// failGrading marks an answer failed after its last grading attempt.
|
||||
func (s *Service) failGrading(ctx context.Context, answerID, attemptID uint, seq int, code string, gradeErr error) error {
|
||||
return lagoon.Transaction(ctx, s.DB, func(ctx context.Context, tx *gorm.DB) error {
|
||||
tx = tx.WithContext(ctx)
|
||||
if _, err := lockAttempt(ctx, tx, attemptID); err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if ok, err := stillPending(ctx, tx, answerID, seq); err != nil || !ok {
|
||||
return err
|
||||
}
|
||||
return tx.Model(&models.AttemptAnswer{}).Where("id = ?", answerID).Updates(map[string]any{
|
||||
"grading_status": models.GradingFailed,
|
||||
"grader": code,
|
||||
"grading_error": cutRunes(gradeErr.Error(), maxGradingError),
|
||||
"updated_at": s.now().UTC(),
|
||||
}).Error
|
||||
})
|
||||
}
|
||||
|
||||
// storeGrade stores a grader's clamped score as graded, recalculates a
|
||||
// completed attempt and fires AnswerGradedEvent after the commit.
|
||||
func (s *Service) storeGrade(ctx context.Context, answerID uint, seq int, attempt *models.Attempt, question models.Question, code string, points float64, feedback string) error {
|
||||
return lagoon.Transaction(ctx, s.DB, func(ctx context.Context, tx *gorm.DB) error {
|
||||
tx = tx.WithContext(ctx)
|
||||
if _, err := lockAttempt(ctx, tx, attempt.ID); err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if ok, err := stillPending(ctx, tx, answerID, seq); err != nil || !ok {
|
||||
return err
|
||||
}
|
||||
feedback = cutRunes(strings.TrimSpace(feedback), maxGraderFeedback)
|
||||
var fb any
|
||||
if feedback != "" {
|
||||
fb = feedback
|
||||
}
|
||||
now := s.now().UTC()
|
||||
if err := tx.Model(&models.AttemptAnswer{}).Where("id = ?", answerID).Updates(map[string]any{
|
||||
"grading_status": models.GradingGraded,
|
||||
"grader": code,
|
||||
"grader_points": points,
|
||||
"points": points,
|
||||
"is_correct": OpenCorrect(points, question.Points),
|
||||
"grader_feedback": fb,
|
||||
"grading_error": nil,
|
||||
"graded_at": now,
|
||||
"updated_at": now,
|
||||
}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.recalcAttempt(ctx, tx, attempt.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
fireAfterCommit(ctx, s, tx, events.AnswerGradedEventName, &events.AnswerGradedEvent{
|
||||
QuizID: attempt.QuizID,
|
||||
QuestionID: question.ID,
|
||||
AttemptID: attempt.ID,
|
||||
AnswerID: answerID,
|
||||
UserID: attempt.UserID,
|
||||
Grader: code,
|
||||
Points: points,
|
||||
MaxPoints: question.Points,
|
||||
Feedback: feedback,
|
||||
})
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// cutRunes keeps at most n runes of s.
|
||||
func cutRunes(s string, n int) string {
|
||||
if utf8.RuneCountInString(s) <= n {
|
||||
return s
|
||||
}
|
||||
r := []rune(s)
|
||||
return string(r[:n])
|
||||
}
|
||||
|
||||
func deref(s *string) string {
|
||||
if s == nil {
|
||||
return ""
|
||||
}
|
||||
return *s
|
||||
}
|
||||
Reference in New Issue
Block a user